When regulars CQL mutations run on Accord use the txn timestamp rather than server timestamp

patch by David Capwell; reviewed by Ariel Weisberg for CASSANDRA-20744
This commit is contained in:
David Capwell 2025-07-30 10:03:47 -07:00
parent 5aeaef5f77
commit 6adac494aa
51 changed files with 2215 additions and 271 deletions

View File

@ -1,4 +1,5 @@
5.1
* When regulars CQL mutations run on Accord use the txn timestamp rather than server timestamp (CASSANDRA-20744)
* When using BEGIN TRANSACTION if a complex mutation exists in the same statement as one that uses a reference, then the complex delete is dropped (CASSANDRA-20788)
* Migrate all nodetool commands from airline to picocli (CASSANDRA-17445)
* Journal.TopologyUpdate should not store the local topology as it can be inferred from the global on (CASSANDRA-20785)

View File

@ -253,10 +253,24 @@ at each replica when `TxnWrite.apply` is called.
`TxnUpdate` is also responsible for populating the update with the
monotonic transactional hybrid logical clock for the execution time of
the transaction. This is used instead of the coordinator generated
timestamp for `SERIAL` and `TransactionStatement` writes. Non-SERIAL
writes use the coordinator or user supplied timestamp although this may
change in between the time of this writing and final release.
the transaction. During migration, normal CQL operations will use the
Accord timestamp once a range starts migration, but will fall back to
server timestamp when migrating away from Accord. For normal CQL operations,
`USING TIMESTAMP` is supported and will cause the data to use the user
timestamp instead of the Accord one, though this breaks linearizability
and should be avoided when possible.
For BATCH operations, timestamp handling is more complex: if the batch
uses `USING TIMESTAMP`, the user timestamp will be used. If all mutations
in a BATCH use `USING TIMESTAMP`, the user timestamp will be used. If not
using `USING TIMESTAMP` and all partition keys are on Accord, the Accord
timestamp is used. If the BATCH has some partitions on Accord and others
not on Accord, the server timestamp will be used (writes to the Accord table will not be linearizable for multi-table
batches where one table is not migrating to Accord). If a batch mixes
server timestamp and `USING TIMESTAMP` mutations, the default behavior
is to reject the batch, configurable via `accord.mixed_time_source_handling`
with values: `reject` (default), `log` (accept and log operations where writes to the Accord table will not be linearizable),
or `ignore` (accept silently).
`TxnUpdate` has a write consistency level that is not visible to Accord
and is it similar to the commit consistency level for CAS writes. If the

View File

@ -279,14 +279,31 @@ Supported read consistency levels are `ONE`, `QUORUM`, `SERIAL`, and
== non-SERIAL consistency
non-SERIAL operations are not linearizable even when executed on Accord
because Accord will continue to write data using the coordinator
generated timestamp not the transaction's timestamp.
non-SERIAL operations are linearizable by default when migrated to Accord,
but during migration they might not be linearizable due to timestamp handling
changes. During migration, normal CQL operations will use the Accord
timestamp once a range starts migration, but will fall back to server
timestamp when migrating away from Accord.
Hints and batch log replay will use server timestamp and they only come into play during
migration as hints and batch log replay won't be generated when a range is on Accord.
`USING TIMESTAMP` is allowed and the application of the operations will
occur in a linearizable order, but from the perspective of a reader the
merged result may not appear linearizable.
For BATCH operations, timestamp handling is more complex:
- If the batch uses `USING TIMESTAMP`, the user timestamp will be used
- If all mutations in a BATCH use `USING TIMESTAMP`, the user timestamp will be used
- If not using `USING TIMESTAMP` and all partition keys are on Accord, the Accord timestamp is used
- If the BATCH has some partitions on Accord and others not on Accord, the server timestamp will be used (writes to the Accord table will not be linearizable for multi-table batches where one table is not migrating to Accord)
- If the batch mixes server timestamp and `USING TIMESTAMP` mutations, the default behavior is to reject the batch, configurable via `accord.mixed_time_source_handling` with values:
* `reject` (default) - reject the CQL operation
* `log` - accept the operation using server timestamp but log operations where writes to the Accord table will not be linearizable
* `ignore` - accept the operation silently
Logged batches that only touch Accord tables will not be written to the batch log because that functionality is redundant with Accord. Batches that touch both Accord and non-Accord tables will be written to the batch log.
Paging runs a separate transaction per page and does not produce a
linearizable result.

View File

@ -73,6 +73,7 @@ import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.PreserveTimestamp;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.WriteResponseHandler;
import org.apache.cassandra.service.accord.IAccordService.IAccordResult;
@ -431,7 +432,7 @@ public class BatchlogManager implements BatchlogManagerMBean
if (accordMutations != null)
{
accordTxnStart = accordTxnStart.withStartedAt(Clock.Global.nanoTime());
accordResult = accordMutations != null ? mutateWithAccordAsync(cm, accordMutations, null, accordTxnStart) : null;
accordResult = accordMutations != null ? mutateWithAccordAsync(cm, accordMutations, null, accordTxnStart, PreserveTimestamp.yes) : null;
}
if (normalMutations != null)

View File

@ -94,7 +94,7 @@ public class AccordSpec
/**
* The queue is backed by submission to a single-threaded plain executor.
* This implementation does not honur the sharding model option.
* This implementation does not honor the sharding model option.
*
* Note: this isn't intended to be used by real clusters.
*/
@ -177,6 +177,13 @@ public class AccordSpec
public boolean state_cache_listener_jfr_enabled = true;
public final JournalSpec journal = new JournalSpec();
public enum MixedTimeSourceHandling
{
reject, log, ignore
}
public volatile MixedTimeSourceHandling mixedTimeSourceHandling = MixedTimeSourceHandling.reject;
public static class JournalSpec implements Params
{
public int segmentSize = 32 << 20;

View File

@ -30,6 +30,8 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.HashMultiset;
import com.google.common.collect.ImmutableList;
@ -71,8 +73,10 @@ import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.ClientWarn;
import org.apache.cassandra.service.PreserveTimestamp;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.TimestampSource;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.transport.Dispatcher;
import org.apache.cassandra.transport.messages.ResultMessage;
@ -198,6 +202,7 @@ public class BatchStatement implements CQLStatement.CompositeCQLStatement
return true;
}
@Override
public void authorize(ClientState state) throws InvalidRequestException, UnauthorizedException
{
for (ModificationStatement statement : statements)
@ -307,7 +312,8 @@ public class BatchStatement implements CQLStatement.CompositeCQLStatement
boolean local,
long batchTimestamp,
long nowInSeconds,
Dispatcher.RequestTime requestTime)
Dispatcher.RequestTime requestTime,
@Nullable TimestampSource.Collector timestampSourceCollector)
{
if (statements.isEmpty())
return Collections.emptyList();
@ -324,6 +330,8 @@ public class BatchStatement implements CQLStatement.CompositeCQLStatement
HashMultiset<ByteBuffer> perKeyCountsForTable = partitionCounts.computeIfAbsent(stmt.metadata.id, k -> HashMultiset.create());
for (int stmtIdx = 0, stmtSize = stmtPartitionKeys.size(); stmtIdx < stmtSize; stmtIdx++)
perKeyCountsForTable.add(stmtPartitionKeys.get(stmtIdx));
if (timestampSourceCollector != null)
timestampSourceCollector.collect(stmt.attrs.isTimestampSet() ? TimestampSource.using : TimestampSource.server);
}
Set<String> tablesWithZeroGcGs = null;
@ -434,6 +442,7 @@ public class BatchStatement implements CQLStatement.CompositeCQLStatement
}
@Override
public ResultMessage execute(QueryState queryState, QueryOptions options, Dispatcher.RequestTime requestTime)
{
return execute(queryState, BatchQueryOptions.withoutPerStatementVariables(options), requestTime);
@ -464,13 +473,29 @@ public class BatchStatement implements CQLStatement.CompositeCQLStatement
if (updatesVirtualTables)
executeInternalWithoutCondition(queryState, options, requestTime);
else
executeWithoutConditions(getMutations(clientState, options, false, timestamp, nowInSeconds, requestTime),
options.getConsistency(), requestTime);
{
PreserveTimestamp preserveTimestamp;
TimestampSource.Collector collector;
if (attrs.isTimestampSet())
{
preserveTimestamp = PreserveTimestamp.yes;
collector = null;
}
else
{
preserveTimestamp = PreserveTimestamp.no;
collector = new TimestampSource.Collector();
}
List<? extends IMutation> mutations = getMutations(clientState, options, false, timestamp, nowInSeconds, requestTime, collector);
if (!mutations.isEmpty() && collector != null)
preserveTimestamp = preserveTimestamp.merge(collector.get());
executeWithoutConditions(mutations, options.getConsistency(), requestTime, preserveTimestamp);
}
return new ResultMessage.Void();
}
private void executeWithoutConditions(List<? extends IMutation> mutations, ConsistencyLevel cl, Dispatcher.RequestTime requestTime) throws RequestExecutionException, RequestValidationException
private void executeWithoutConditions(List<? extends IMutation> mutations, ConsistencyLevel cl, Dispatcher.RequestTime requestTime, PreserveTimestamp preserveTimestamp) throws RequestExecutionException, RequestValidationException
{
if (mutations.isEmpty())
return;
@ -481,7 +506,7 @@ public class BatchStatement implements CQLStatement.CompositeCQLStatement
updatePartitionsPerBatchMetrics(mutations.size());
boolean mutateAtomic = (isLogged() && mutations.size() > 1);
StorageProxy.mutateWithTriggers(mutations, cl, mutateAtomic, requestTime);
StorageProxy.mutateWithTriggers(mutations, cl, mutateAtomic, requestTime, preserveTimestamp);
ClientRequestSizeMetrics.recordRowAndColumnCountMetrics(mutations);
}
@ -611,7 +636,7 @@ public class BatchStatement implements CQLStatement.CompositeCQLStatement
long timestamp = batchOptions.getTimestamp(queryState);
long nowInSeconds = batchOptions.getNowInSeconds(queryState);
for (IMutation mutation : getMutations(queryState.getClientState(), batchOptions, true, timestamp, nowInSeconds, requestTime))
for (IMutation mutation : getMutations(queryState.getClientState(), batchOptions, true, timestamp, nowInSeconds, requestTime, null))
mutation.apply();
return null;
}

View File

@ -62,6 +62,7 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableParams;
import org.apache.cassandra.service.CASRequest;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.PreserveTimestamp;
import org.apache.cassandra.service.accord.api.PartitionKey;
import org.apache.cassandra.service.accord.serializers.TableMetadatas;
import org.apache.cassandra.service.accord.serializers.TableMetadatasAndKeys;
@ -528,7 +529,7 @@ public class CQL3CasRequest implements CASRequest
TableParams tableParams = tableMetadata.params;
commitConsistencyLevel = tableParams.transactionalMode.commitCLForMode(tableParams.transactionalMigrationFrom, commitConsistencyLevel, cm, tableMetadata.id, key.getToken());
// CAS requires using the new txn timestamp to correctly linearize some kinds of updates
return new TxnUpdate(tables, createWriteFragments(clientState), createCondition(), commitConsistencyLevel, false);
return new TxnUpdate(tables, createWriteFragments(clientState), createCondition(), commitConsistencyLevel, PreserveTimestamp.no);
}
private TxnCondition createCondition()

View File

@ -109,6 +109,7 @@ import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.ViewMetadata;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.PreserveTimestamp;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.accord.api.PartitionKey;
@ -364,6 +365,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
return attrs.getTimeToLive(options, metadata);
}
@Override
public void authorize(ClientState state) throws InvalidRequestException, UnauthorizedException
{
state.ensureTablePermission(metadata, Permission.MODIFY);
@ -648,7 +650,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
);
if (!mutations.isEmpty())
{
StorageProxy.mutateWithTriggers(mutations, cl, false, requestTime);
StorageProxy.mutateWithTriggers(mutations, cl, false, requestTime, attrs.isTimestampSet() ? PreserveTimestamp.yes : PreserveTimestamp.no);
if (!SchemaConstants.isSystemKeyspace(metadata.keyspace))
ClientRequestSizeMetrics.recordRowAndColumnCountMetrics(mutations);

View File

@ -66,6 +66,7 @@ import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableParams;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.PreserveTimestamp;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.api.PartitionKey;
@ -469,7 +470,7 @@ public class TransactionStatement implements CQLStatement.CompositeCQLStatement,
ConsistencyLevel commitCL = consistencyLevelForAccordCommit(cm, tables, keyCollector, options.getConsistency());
List<TxnNamedRead> reads = createNamedReads(options, autoReads, keyCollector);
Keys keys = keyCollector.build();
AccordUpdate update = new TxnUpdate(tables, writeFragments, createCondition(options), commitCL, false);
AccordUpdate update = new TxnUpdate(tables, writeFragments, createCondition(options), commitCL, PreserveTimestamp.no);
TxnRead read = createTxnRead(tables, reads, null, Domain.Key);
return new Txn.InMemory(keys, read, TxnQuery.ALL, update, new TableMetadatasAndKeys(tables, keys));
}

View File

@ -2229,6 +2229,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
CompactionManager.instance.performMaximal(this, splitOutput, permittedParallelism);
}
@VisibleForTesting
public List<Future<?>> submitMajorCompaction(boolean splitOutput, int permittedParallelism)
{
return CompactionManager.instance.submitMaximal(this, splitOutput, permittedParallelism);
}
@Override
public void forceCompactionForTokenRange(Collection<Range<Token>> tokenRanges) throws ExecutionException, InterruptedException
{

View File

@ -1141,6 +1141,11 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan
FBUtilities.waitOnFutures(submitMaximal(cfStore, getDefaultGcBefore(cfStore, FBUtilities.nowInSeconds()), splitOutput, permittedParallelism));
}
public List<Future<?>> submitMaximal(final ColumnFamilyStore cfStore, boolean splitOutput, int permittedParallelism)
{
return submitMaximal(cfStore, getDefaultGcBefore(cfStore, FBUtilities.nowInSeconds()), splitOutput, permittedParallelism);
}
public List<Future<?>> submitMaximal(final ColumnFamilyStore cfStore, final long gcBefore, boolean splitOutput, int permittedParallelism)
{
return submitMaximal(cfStore, gcBefore, splitOutput, permittedParallelism, OperationType.MAJOR_COMPACTION);

View File

@ -51,6 +51,7 @@ import org.apache.cassandra.metrics.HintsServiceMetrics;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.RequestCallback;
import org.apache.cassandra.service.PreserveTimestamp;
import org.apache.cassandra.service.accord.IAccordService.IAccordResult;
import org.apache.cassandra.service.accord.txn.TxnResult;
import org.apache.cassandra.service.consensus.migration.ConsensusMigrationMutationHelper;
@ -355,7 +356,7 @@ final class HintsDispatcher implements AutoCloseable
if (accordHintMutation != null)
{
requestTime = Dispatcher.RequestTime.forImmediateExecution();
accordTxnResult = accordHintMutation != null ? ConsensusMigrationMutationHelper.instance().mutateWithAccordAsync(cm, accordHintMutation, null, requestTime) : null;
accordTxnResult = accordHintMutation != null ? ConsensusMigrationMutationHelper.instance().mutateWithAccordAsync(cm, accordHintMutation, null, requestTime, PreserveTimestamp.yes) : null;
}
Hint normalHint = splitHint.normalHint;

View File

@ -0,0 +1,59 @@
/*
* 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.service;
/**
* When running on Accord each cell's timestamp is updated to be the transaction timestamp, but some use cases need a way
* to opt out of this behavior and preserve the timestamp; examples are hints and batches that have some mutations using
* the server's timestamp and others using USING TIMESTAMP.
*/
public enum PreserveTimestamp
{
yes(true),
/**
* When working with a BATCH, different mutations can have different time sources (one mutation might be
* server based, and another is USING TIMESTAMP based); "mixed" is for this case.
*/
mixedTimeSource(true),
no(false),
;
public final boolean preserve;
PreserveTimestamp(boolean preserve)
{
this.preserve = preserve;
}
public static PreserveTimestamp merge(PreserveTimestamp left, TimestampSource right)
{
if (right == TimestampSource.mixed)
return mixedTimeSource;
if (left.preserve)
return left;
if (right == TimestampSource.using)
return yes;
return left;
}
public PreserveTimestamp merge(TimestampSource right)
{
return merge(this, right);
}
}

View File

@ -58,6 +58,7 @@ import org.apache.cassandra.batchlog.Batch;
import org.apache.cassandra.batchlog.BatchlogManager;
import org.apache.cassandra.concurrent.DebuggableTask.RunnableDebuggableTask;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.config.AccordSpec;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
@ -241,6 +242,7 @@ public class StorageProxy implements StorageProxyMBean
public static final String UNREACHABLE = "UNREACHABLE";
private static final int FAILURE_LOGGING_INTERVAL_SECONDS = CassandraRelevantProperties.FAILURE_LOGGING_INTERVAL_SECONDS.getInt();
private static final String UNSAFE_MIXED_MUTATIONS_MSG = "Mutations look to have different time sources, some are using 'USING TIMESTAMP' and others are using the server timestamp; writes to the Accord table will not be linearizable while using transactions. To allow this behavior set accord.mixed_time_source_handling=log or ignore";
private static final WritePerformer standardWritePerformer;
private static final WritePerformer counterWritePerformer;
@ -1215,7 +1217,8 @@ public class StorageProxy implements StorageProxyMBean
public static void mutateWithTriggers(List<? extends IMutation> mutations,
ConsistencyLevel consistencyLevel,
boolean mutateAtomically,
Dispatcher.RequestTime requestTime)
Dispatcher.RequestTime requestTime,
PreserveTimestamp preserveTimestamps)
throws WriteTimeoutException, WriteFailureException, UnavailableException, OverloadedException, InvalidRequestException
{
if (DatabaseDescriptor.getPartitionDenylistEnabled() && DatabaseDescriptor.getDenylistWritesEnabled())
@ -1250,20 +1253,30 @@ public class StorageProxy implements StorageProxyMBean
if (augmented != null || mutateAtomically || updatesView)
mutateAtomically(augmented != null ? augmented : (List<Mutation>)mutations, consistencyLevel, updatesView, requestTime);
else
dispatchMutationsWithRetryOnDifferentSystem(mutations, consistencyLevel, requestTime);
dispatchMutationsWithRetryOnDifferentSystem(mutations, consistencyLevel, requestTime, preserveTimestamps);
}
public static void dispatchMutationsWithRetryOnDifferentSystem(List<? extends IMutation> mutations, ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime)
public static void dispatchMutationsWithRetryOnDifferentSystem(List<? extends IMutation> mutations, ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime, PreserveTimestamp preserveTimestamps)
{
while (true)
{
ClusterMetadata cm = ClusterMetadata.current();
try
{
SplitMutations splitMutations = splitMutationsIntoAccordAndNormal(cm, (List<IMutation>)mutations);
SplitMutations<?> splitMutations = splitMutationsIntoAccordAndNormal(cm, (List<IMutation>)mutations);
List<? extends IMutation> accordMutations = splitMutations.accordMutations();
IAccordResult<TxnResult> accordResult = accordMutations != null ? mutateWithAccordAsync(cm, accordMutations, consistencyLevel, requestTime) : null;
List<? extends IMutation> normalMutations = splitMutations.normalMutations();
// If there was ever any attempt to apply part of the mutation using the eventually consistent path
// then we need to continue to use the timestamp used by the eventually consistent path to not
// end up with multiple timestamps, but if it only ever used the transactional path then we can
// use the transactional timestamp to get linearizability
if (!preserveTimestamps.preserve && normalMutations != null)
preserveTimestamps = PreserveTimestamp.yes;
// A BATCH statement has multiple mutations mixing server timestamps and `USING TIMESTAMP`,
// which is not linearizable for the writes to Accord tables.
if (accordMutations != null && preserveTimestamps == PreserveTimestamp.mixedTimeSource)
checkMixedTimeSourceHandling();
IAccordResult<TxnResult> accordResult = accordMutations != null ? mutateWithAccordAsync(cm, accordMutations, consistencyLevel, requestTime, preserveTimestamps) : null;
Tracing.trace("Split mutations into Accord {} and normal {}", accordMutations, normalMutations);
Throwable failure = null;
@ -1330,6 +1343,26 @@ public class StorageProxy implements StorageProxyMBean
}
}
private static void checkMixedTimeSourceHandling()
{
AccordSpec.MixedTimeSourceHandling handling = DatabaseDescriptor.getAccord().mixedTimeSourceHandling;
switch (handling)
{
case log:
case reject:
{
ClientWarn.instance.warn(UNSAFE_MIXED_MUTATIONS_MSG);
logger.warn(UNSAFE_MIXED_MUTATIONS_MSG);
if (handling == AccordSpec.MixedTimeSourceHandling.reject)
throw new InvalidRequestException(UNSAFE_MIXED_MUTATIONS_MSG);
}
break;
case ignore:
// ignore
break;
}
}
private static ConsistencyLevel consistencyLevelForBatchLog(ConsistencyLevel consistencyLevel, boolean requireQuorumForRemove)
{
// If we are requiring quorum nodes for removal, we upgrade consistency level to QUORUM unless we already
@ -1467,7 +1500,7 @@ public class StorageProxy implements StorageProxyMBean
}
// Start Accord executing so it executes while the mutations are synchronously applied
IAccordResult<TxnResult> accordResult = !accordMutations.isEmpty() ? mutateWithAccordAsync(cm, accordMutations, consistencyLevel, requestTime) : null;
IAccordResult<TxnResult> accordResult = !accordMutations.isEmpty() ? mutateWithAccordAsync(cm, accordMutations, consistencyLevel, requestTime, PreserveTimestamp.yes) : null;
Throwable failure = null;
try

View File

@ -0,0 +1,99 @@
/*
* 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.service;
/**
* Represents where a timestamp is coming from.
*/
public enum TimestampSource
{
unknown, server, using,
/**
* When doing a BATCH some mutations use "server" and others use "using".
*/
mixed;
public static TimestampSource merge(TimestampSource left, TimestampSource right)
{
if (left == right) return left;
switch (left)
{
case unknown:
return right;
case server:
if (right == TimestampSource.unknown)
{
return server;
}
else if (right == TimestampSource.using || right == TimestampSource.mixed)
{
return mixed;
}
throw new UnsupportedOperationException(right.name());
case using:
if (right == TimestampSource.unknown)
{
return using;
}
else if (right == TimestampSource.server || right == TimestampSource.mixed)
{
return mixed;
}
throw new UnsupportedOperationException(right.name());
case mixed:
return left;
default:
throw new UnsupportedOperationException(left.name());
}
}
public static class Collector
{
boolean hasServerTimestamp = false;
boolean hasUserTimestamp = false;
public void collect(TimestampSource source)
{
switch (source)
{
case unknown:
case server:
hasServerTimestamp = true;
break;
case using:
hasUserTimestamp = true;
break;
case mixed:
hasServerTimestamp = true;
hasUserTimestamp = true;
break;
default:
throw new UnsupportedOperationException(source.name());
}
}
public TimestampSource get()
{
if (hasServerTimestamp && hasUserTimestamp) return mixed;
if (hasServerTimestamp) return server;
if (hasUserTimestamp) return using;
throw new IllegalStateException(".get() called before .collect()");
}
}
}

View File

@ -44,6 +44,7 @@ import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.service.PreserveTimestamp;
import org.apache.cassandra.service.accord.AccordObjectSizes;
import org.apache.cassandra.service.accord.IAccordService;
import org.apache.cassandra.service.accord.api.PartitionKey;
@ -74,7 +75,7 @@ import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSi
public class TxnUpdate extends AccordUpdate
{
private static final long EMPTY_SIZE = ObjectSizes.measure(new TxnUpdate(TableMetadatas.none(), null, new ByteBuffer[0], null, null, false));
private static final long EMPTY_SIZE = ObjectSizes.measure(new TxnUpdate(TableMetadatas.none(), null, new ByteBuffer[0], null, null, PreserveTimestamp.no));
private static final int FLAG_PRESERVE_TIMESTAMPS = 0x1;
final TableMetadatas tables;
@ -88,12 +89,12 @@ public class TxnUpdate extends AccordUpdate
// Hints and batchlog want to write with the lower timestamp they generated when applying their writes via Accord
// so they don't resurrect data if they are applied at a later time. Accord should be fine with this because
// the writes are still deterministic from the perspective of coordinators/recovery coordinators.
private final boolean preserveTimestamps;
private final PreserveTimestamp preserveTimestamps;
// Memoize computation of condition
private Boolean conditionResult;
public TxnUpdate(TableMetadatas tables, List<Fragment> fragments, TxnCondition condition, @Nullable ConsistencyLevel cassandraCommitCL, boolean preserveTimestamps)
public TxnUpdate(TableMetadatas tables, List<Fragment> fragments, TxnCondition condition, @Nullable ConsistencyLevel cassandraCommitCL, PreserveTimestamp preserveTimestamps)
{
requireArgument(cassandraCommitCL == null || IAccordService.SUPPORTED_COMMIT_CONSISTENCY_LEVELS.contains(cassandraCommitCL));
this.tables = tables;
@ -110,7 +111,7 @@ public class TxnUpdate extends AccordUpdate
this.preserveTimestamps = preserveTimestamps;
}
private TxnUpdate(TableMetadatas tables, Keys keys, ByteBuffer[] fragments, SerializedTxnCondition condition, ConsistencyLevel cassandraCommitCL, boolean preserveTimestamps)
private TxnUpdate(TableMetadatas tables, Keys keys, ByteBuffer[] fragments, SerializedTxnCondition condition, ConsistencyLevel cassandraCommitCL, PreserveTimestamp preserveTimestamps)
{
this.tables = tables;
this.keys = keys;
@ -122,7 +123,7 @@ public class TxnUpdate extends AccordUpdate
public static TxnUpdate empty()
{
return new TxnUpdate(TableMetadatas.none(), Collections.emptyList(), TxnCondition.none(), null, false);
return new TxnUpdate(TableMetadatas.none(), Collections.emptyList(), TxnCondition.none(), null, PreserveTimestamp.no);
}
@Override
@ -168,7 +169,7 @@ public class TxnUpdate extends AccordUpdate
// Batch log and hints want to keep their lower timestamp for the applied writes to avoid resurrecting old data
// when they are applied later, possibly after further updates have already been acknowledged.
public boolean preserveTimestamps()
public PreserveTimestamp preserveTimestamps()
{
return preserveTimestamps;
}
@ -271,7 +272,8 @@ public class TxnUpdate extends AccordUpdate
{
// Serializing it with the condition result set shouldn't be needed
checkState(update.conditionResult == null, "Can't serialize if conditionResult is set without adding it to serialization");
out.writeByte(update.preserveTimestamps ? FLAG_PRESERVE_TIMESTAMPS : 0);
// Once in accord "mixedTimeSource" and "yes" are the same, so only care about the side effect: that the timestamp is preserved or not
out.writeByte(update.preserveTimestamps.preserve ? FLAG_PRESERVE_TIMESTAMPS : 0);
tablesAndKeys.serializeKeys(update.keys, out);
writeWithVIntLength(update.condition.bytes(), out);
serializeArray(update.fragments, out, ByteBufferUtil.byteBufferSerializer);
@ -287,7 +289,7 @@ public class TxnUpdate extends AccordUpdate
ByteBuffer condition = readWithVIntLength(in);
ByteBuffer[] fragments = deserializeArray(in, ByteBufferUtil.byteBufferSerializer, ByteBuffer[]::new);
ConsistencyLevel consistencyLevel = deserializeNullable(in, consistencyLevelSerializer);
return new TxnUpdate(tablesAndKeys.tables, keys, fragments, new SerializedTxnCondition(condition), consistencyLevel, preserveTimestamps);
return new TxnUpdate(tablesAndKeys.tables, keys, fragments, new SerializedTxnCondition(condition), consistencyLevel, preserveTimestamps ? PreserveTimestamp.yes : PreserveTimestamp.no);
}
@Override

View File

@ -447,7 +447,7 @@ public class TxnWrite extends AbstractKeySorted<TxnWrite.Update> implements Writ
if (isConditionMet)
{
AccordExecutor executor = ((AccordCommandStore) commandStore).executor();
boolean preserveTimestamps = txnUpdate.preserveTimestamps();
boolean preserveTimestamps = txnUpdate.preserveTimestamps().preserve;
// Apply updates not specified fully by the client but built from fragments completed by data from reads.
// This occurs, for example, when an UPDATE statement uses a value assigned by a LET statement.
forEachWithKey(key, write -> results.add(write.write(executor, tables, preserveTimestamps, timestamp)));

View File

@ -48,6 +48,7 @@ import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableParams;
import org.apache.cassandra.service.PreserveTimestamp;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.IAccordService;
import org.apache.cassandra.service.accord.IAccordService.IAccordResult;
@ -234,12 +235,16 @@ public class ConsensusMigrationMutationHelper
return new SplitMutation(accordMutation, normalMutation);
}
public IAccordResult<TxnResult> mutateWithAccordAsync(ClusterMetadata cm, Mutation mutation, @Nullable ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime)
public IAccordResult<TxnResult> mutateWithAccordAsync(ClusterMetadata cm, Mutation mutation, @Nullable ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime, PreserveTimestamp preserveTimestamps)
{
return mutateWithAccordAsync(cm, ImmutableList.of(mutation), consistencyLevel, requestTime);
return mutateWithAccordAsync(cm, ImmutableList.of(mutation), consistencyLevel, requestTime, preserveTimestamps);
}
public static IAccordResult<TxnResult> mutateWithAccordAsync(ClusterMetadata cm, Collection<? extends IMutation> mutations, @Nullable ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime)
public static IAccordResult<TxnResult> mutateWithAccordAsync(ClusterMetadata cm,
Collection<? extends IMutation> mutations,
@Nullable ConsistencyLevel consistencyLevel,
Dispatcher.RequestTime requestTime,
PreserveTimestamp preserveTimestamps)
{
if (consistencyLevel != null && !IAccordService.SUPPORTED_COMMIT_CONSISTENCY_LEVELS.contains(consistencyLevel))
throw new InvalidRequestException(consistencyLevel + " is not supported by Accord");
@ -272,7 +277,7 @@ public class ConsensusMigrationMutationHelper
// Potentially ignore commit consistency level if the TransactionalMode specifies full
ConsistencyLevel clForCommit = consistencyLevelForCommit(cm, mutations, consistencyLevel);
TableMetadatasAndKeys tablesAndKeys = new TableMetadatasAndKeys(tables, keyCollector.build());
TxnUpdate update = new TxnUpdate(tables, fragments, TxnCondition.none(), clForCommit, true);
TxnUpdate update = new TxnUpdate(tables, fragments, TxnCondition.none(), clForCommit, preserveTimestamps);
Txn.InMemory txn = new Txn.InMemory(tablesAndKeys.keys, TxnRead.empty(Domain.Key), TxnQuery.NONE, update, tablesAndKeys);
return AccordService.instance().coordinateAsync(minEpoch, txn, clForCommit, requestTime);
}

View File

@ -43,14 +43,21 @@ public class Query implements IIsolatedExecutor.SerializableCallable<SimpleQuery
public final String query;
final long timestamp;
final boolean deserializeResult;
final org.apache.cassandra.distributed.api.ConsistencyLevel commitConsistencyOrigin;
final org.apache.cassandra.distributed.api.ConsistencyLevel serialConsistencyOrigin;
public final Object[] boundValues;
public Query(String query, long timestamp, org.apache.cassandra.distributed.api.ConsistencyLevel commitConsistencyOrigin, org.apache.cassandra.distributed.api.ConsistencyLevel serialConsistencyOrigin, Object[] boundValues)
{
this(query, timestamp, true, commitConsistencyOrigin, serialConsistencyOrigin, boundValues);
}
public Query(String query, long timestamp, boolean deserializeResult, org.apache.cassandra.distributed.api.ConsistencyLevel commitConsistencyOrigin, org.apache.cassandra.distributed.api.ConsistencyLevel serialConsistencyOrigin, Object[] boundValues)
{
this.query = query;
this.timestamp = timestamp;
this.deserializeResult = deserializeResult;
this.commitConsistencyOrigin = commitConsistencyOrigin;
this.serialConsistencyOrigin = serialConsistencyOrigin;
this.boundValues = boundValues;
@ -90,7 +97,7 @@ public class Query implements IIsolatedExecutor.SerializableCallable<SimpleQuery
if (res != null)
res.setWarnings(ClientWarn.instance.getWarnings());
return RowUtil.toQueryResult(res);
return RowUtil.toQueryResult(res, deserializeResult);
}
public String toString()

View File

@ -37,13 +37,20 @@ import org.apache.cassandra.transport.messages.ResultMessage;
public class RowUtil
{
private static final ByteBuffer[][] EMPTY_BB_ROWS = new ByteBuffer[0][];
public static SimpleQueryResult toQueryResult(ResultMessage res)
{
return toQueryResult(res, true);
}
public static SimpleQueryResult toQueryResult(ResultMessage res, boolean deserialize)
{
if (res != null && res.kind == ResultMessage.Kind.ROWS)
{
ResultMessage.Rows rows = (ResultMessage.Rows) res;
String[] names = getColumnNames(rows.result.metadata.requestNames());
Object[][] results = toObjects(rows);
Object[][] results = toObjects(rows, deserialize);
// Warnings may be null here, due to ClientWarn#getWarnings() handling of empty warning lists.
List<String> warnings = res.getWarnings();
@ -67,12 +74,17 @@ public class RowUtil
return names.stream().map(c -> c.name.toString()).toArray(String[]::new);
}
public static Object[][] toObjects(ResultMessage.Rows rows)
public static Object[][] toObjects(ResultMessage.Rows rows, boolean deserialize)
{
return toObjects(rows.result.metadata.requestNames(), rows.result.rows);
return toObjects(rows.result.metadata.requestNames(), rows.result.rows, deserialize);
}
public static Object[][] toObjects(List<ColumnSpecification> specs, List<List<ByteBuffer>> rows)
{
return toObjects(specs, rows, true);
}
public static Object[][] toObjects(List<ColumnSpecification> specs, List<List<ByteBuffer>> rows, boolean deserialize)
{
Object[][] result = new Object[rows.size()][];
for (int i = 0; i < rows.size(); i++)
@ -84,12 +96,34 @@ public class RowUtil
ByteBuffer bb = row.get(j);
if (bb != null)
result[i][j] = specs.get(j).type.getSerializer().deserialize(bb);
result[i][j] = deserialize ? specs.get(j).type.getSerializer().deserialize(bb) : bb;
}
}
return result;
}
/**
* When {@code deserialize=false} then all values are {@link ByteBuffer}, so this function can be used to convert the {@code Object[][]} to
* its {@link ByteBuffer} form.
*
* Calling this function if {@code deserialize=true} should be expected to fail, but should not be relied on; it is
* safer to assume the behavior is undefined in this case.
*/
public static ByteBuffer[][] toByteBuffer(Object[][] rows)
{
if (rows.length == 0) return EMPTY_BB_ROWS;
ByteBuffer[][] result = new ByteBuffer[rows.length][];
for (int i = 0; i < rows.length; i++)
{
Object[] in = rows[i];
ByteBuffer[] out = new ByteBuffer[in.length];
for (int j = 0; j < in.length; j++)
out[j] = (ByteBuffer) in[j];
result[i] = out;
}
return result;
}
public static Iterator<Object[]> toObjects(ResultSet rs)
{
return Iterators.transform(rs.iterator(), (Row row) -> {
@ -116,15 +150,15 @@ public class RowUtil
public static Iterator<Object[]> toIter(List<ColumnSpecification> columnSpecs, Iterator<UntypedResultSet.Row> rs)
{
Iterator<List<ByteBuffer>> iter = Iterators.transform(rs,
(row) -> {
List<ByteBuffer> bbs = new ArrayList<>(columnSpecs.size());
for (int i = 0; i < columnSpecs.size(); i++)
{
ColumnSpecification columnSpec = columnSpecs.get(i);
bbs.add(row.getBytes(columnSpec.name.toString()));
}
return bbs;
});
(row) -> {
List<ByteBuffer> bbs = new ArrayList<>(columnSpecs.size());
for (int i = 0; i < columnSpecs.size(); i++)
{
ColumnSpecification columnSpec = columnSpecs.get(i);
bbs.add(row.getBytes(columnSpec.name.toString()));
}
return bbs;
});
return toIterInternal(columnSpecs, Lists.newArrayList(iter));
}

View File

@ -48,6 +48,8 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.Futures;
import org.agrona.collections.IntArrayList;
import org.apache.cassandra.tcm.compatibility.TokenRingUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Shared;
@ -686,10 +688,21 @@ public class ClusterUtils
waitForCMSToQuiesce(cluster, maxEpoch(cluster, cmsNodes));
}
public static Epoch maxEpoch(ICluster<IInvokableInstance> cluster, int[] cmsNodes)
public static Epoch maxEpoch(ICluster<IInvokableInstance> cluster)
{
IntArrayList up = new IntArrayList();
for (var inst : cluster)
{
if (inst.isShutdown()) continue;
up.add(inst.config().num());
}
return maxEpoch(cluster, up.toIntArray());
}
public static Epoch maxEpoch(ICluster<IInvokableInstance> cluster, int[] nodes)
{
Epoch max = null;
for (int id : cmsNodes)
for (int id : nodes)
{
IInvokableInstance inst = cluster.get(id);
if (inst.isShutdown()) continue;
@ -698,7 +711,7 @@ public class ClusterUtils
max = version;
}
if (max == null)
throw new AssertionError("Unable to find max epoch from " + cmsNodes);
throw new AssertionError("Unable to find max epoch from " + Arrays.toString(nodes));
return max;
}

View File

@ -23,9 +23,11 @@ import accord.utils.RandomSource;
import org.apache.cassandra.cql3.KnownIssue;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
import org.apache.cassandra.tcm.Epoch;
public abstract class AccordInteropMultiNodeTableWalkBase extends MultiNodeTableWalkBase
@ -83,9 +85,27 @@ Suppressed: java.lang.AssertionError: Unknown keyspace ks12
public class AccordInteropMultiNodeState extends MultiNodeState
{
private final boolean allowUsingTimestamp;
public AccordInteropMultiNodeState(RandomSource rs, Cluster cluster)
{
super(rs, cluster);
allowUsingTimestamp = rs.nextBoolean();
}
@Override
protected void createTable(TableMetadata metadata)
{
super.createTable(metadata);
Epoch maxEpoch = ClusterUtils.maxEpoch(cluster);
ClusterUtils.waitForCMSToQuiesce(cluster, maxEpoch);
ClusterUtils.awaitAccordEpochReady(cluster, maxEpoch.getEpoch());
}
@Override
protected boolean allowUsingTimestamp()
{
return allowUsingTimestamp;
}
@Override

View File

@ -367,6 +367,7 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
statefulBuilder.check(commands(() -> rs -> createState(rs, cluster))
.add(StatefulASTBase::insert)
.add(StatefulASTBase::fullTableScan)
.addIf(State::allowUsingTimestamp, StatefulASTBase::validateUsingTimestamp)
.addIf(State::hasPartitions, this::selectExisting)
.addAllIf(State::supportTokens,
this::selectToken,

View File

@ -268,6 +268,7 @@ public class SingleNodeTokenConflictTest extends StatefulASTBase
preCheck(cluster, statefulBuilder);
statefulBuilder.check(commands(() -> rs -> createState(rs, cluster))
.add(StatefulASTBase::insert)
.addIf(State::allowUsingTimestamp, StatefulASTBase::validateUsingTimestamp)
//TODO (now, coverage): this is flakey and non-deterministic. When this fails (gives bad response) rerunning the seed yields a passing test!
// .add(StatefulASTBase::fullTableScan)
.add(SingleNodeTokenConflictTest::pkEq)

View File

@ -69,6 +69,8 @@ import org.apache.cassandra.cql3.ast.Visitor.CompositeVisitor;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.AsciiType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.ListType;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.distributed.Cluster;
@ -90,8 +92,10 @@ import org.apache.cassandra.utils.AbstractTypeGenerators;
import org.apache.cassandra.utils.CassandraGenerators;
import org.apache.cassandra.utils.FastByteOperations;
import org.apache.cassandra.utils.Generators;
import org.assertj.core.api.Assertions;
import org.quicktheories.generators.SourceDSL;
import static accord.utils.Property.ignoreCommand;
import static accord.utils.Property.multistep;
import static org.apache.cassandra.distributed.test.JavaDriverUtils.toDriverCL;
import static org.apache.cassandra.utils.AbstractTypeGenerators.overridePrimitiveTypeSupport;
@ -125,6 +129,7 @@ public class StatefulASTBase extends TestBaseImpl
protected static final Gen<Gen.IntGen> LIMIT_DISTRO = Gens.mixedDistribution(1, 1001);
protected static final Gen<Gen.IntGen> REPAIR_TYPE_EMPTY_MODEL_DISTRO = Gens.mixedDistribution(0, 2);
protected static final Gen<Gen.IntGen> REPAIR_TYPE_DISTRO = Gens.mixedDistribution(0, 3);
private static final ListType<Long> LONG_LIST_TYPE = ListType.getInstance(LongType.instance, false);
static
{
@ -206,10 +211,55 @@ public class StatefulASTBase extends TestBaseImpl
});
}
protected static <S extends CommonState> Property.Command<S, Void, ?> validateUsingTimestamp(RandomSource rs, S state)
{
if (state.operations == 0)
return ignoreCommand();
var builder = Select.builder(state.metadata);
for (var c : state.model.factory.regularAndStaticColumns)
builder.selection(FunctionCall.writetime(c));
ByteBuffer upperboundTimestamp = LongType.instance.decompose((long) state.operations);
var select = builder.build();
var inst = state.selectInstance(rs);
return new Property.SimpleCommand<>(state.humanReadable(select, null), s -> {
var result = s.executeQuery(inst, Integer.MAX_VALUE, s.selectCl(), select);
for (var row : result)
{
for (var col : state.model.factory.regularAndStaticColumns)
{
int idx = state.model.factory.regularAndStaticColumns.indexOf(col);
ByteBuffer value = row[idx];
if (value == null) continue;
if (col.type().isMultiCell())
{
List<ByteBuffer> timestamps = LONG_LIST_TYPE.unpack(value);
int cellIndex = 0;
for (var timestamp : timestamps)
{
Assertions.assertThat(LongType.instance.compare(timestamp, upperboundTimestamp))
.describedAs("Unexected timestamp at multi-cell index %s for col %s: %s > %s", cellIndex, col, LongType.instance.compose(timestamp), state.operations)
.isLessThanOrEqualTo(state.operations);
cellIndex++;
}
}
else
{
Assertions.assertThat(LongType.instance.compare(value, upperboundTimestamp))
.describedAs("Unexected timestamp for col %s: %s > %s", col, LongType.instance.compose(value), state.operations)
.isLessThanOrEqualTo(state.operations);
}
}
}
});
}
protected static <S extends CommonState> Property.Command<S, Void, ?> insert(RandomSource rs, S state)
{
int timestamp = ++state.operations;
Mutation mutation = state.mutationGen().next(rs).withTimestamp(timestamp);
Mutation original = state.mutationGen().next(rs);
Mutation mutation = state.allowUsingTimestamp()
? original.withTimestamp(timestamp)
: original;
if (!state.readAfterWrite())
return state.command(rs, mutation);
@ -440,6 +490,11 @@ public class StatefulASTBase extends TestBaseImpl
return false;
}
protected boolean allowUsingTimestamp()
{
return true;
}
protected RepairGenerators.Builder repairArgsBuilder()
{
return new RepairGenerators.Builder(i -> Arrays.asList(metadata.keyspace, metadata.name))
@ -709,7 +764,7 @@ public class StatefulASTBase extends TestBaseImpl
return ret.toArray(a);
}
private String humanReadable(Statement stmt, @Nullable String annotate)
protected String humanReadable(Statement stmt, @Nullable String annotate)
{
// With UTF-8 some chars can cause printing issues leading to error messages that don't reproduce the original issue.
// To avoid this problem, always escape the CQL so nothing gets lost

View File

@ -64,6 +64,7 @@ import org.apache.cassandra.cql3.ast.ReferenceExpression;
import org.apache.cassandra.cql3.ast.Select;
import org.apache.cassandra.cql3.ast.StandardVisitors;
import org.apache.cassandra.cql3.ast.Symbol;
import org.apache.cassandra.cql3.ast.Txn;
import org.apache.cassandra.cql3.ast.Value;
import org.apache.cassandra.cql3.ast.Visitor;
import org.apache.cassandra.db.BufferClustering;
@ -83,10 +84,26 @@ import org.apache.cassandra.utils.ImmutableUniqueList;
import org.apache.cassandra.utils.Pair;
import static org.apache.cassandra.cql3.ast.Elements.symbols;
import static org.apache.cassandra.harry.MagicConstants.NO_TIMESTAMP;
import static org.apache.cassandra.harry.model.BytesPartitionState.asCQL;
public class ASTSingleTableModel
{
private enum CasResponse
{
full, applied, none;
boolean validate()
{
switch (this)
{
case full:
case applied:
return true;
}
return false;
}
}
private static final ByteBuffer[][] NO_ROWS = new ByteBuffer[0][];
private static final Symbol CAS_APPLIED = new Symbol.UnquotedSymbol("[applied]", BooleanType.instance);
private static final ImmutableUniqueList<Symbol> CAS_APPLIED_COLUMNS = ImmutableUniqueList.<Symbol>builder().add(CAS_APPLIED).build();
@ -98,6 +115,7 @@ public class ASTSingleTableModel
private final EnumSet<KnownIssue> ignoredIssues;
private final TreeMap<BytesPartitionState.Ref, BytesPartitionState> partitions = new TreeMap<>();
private long numMutations = 0;
private CasResponse validateCass = CasResponse.full;
public ASTSingleTableModel(TableMetadata metadata)
{
@ -110,6 +128,11 @@ public class ASTSingleTableModel
this.ignoredIssues = Objects.requireNonNull(ignoredIssues);
}
public void validateCasAppliedOnly()
{
validateCass = CasResponse.applied;
}
public NavigableSet<BytesPartitionState.Ref> partitionKeys()
{
return partitions.navigableKeySet();
@ -209,6 +232,86 @@ public class ASTSingleTableModel
}
}
public boolean isConditional(Txn txn)
{
return txn.ifBlock.isPresent();
}
public boolean isReadOnly(Txn txn)
{
return txn.ifBlock.isEmpty() && txn.mutations.isEmpty();
}
public boolean shouldApply(Txn txn)
{
if (!txn.ifBlock.isPresent()) return true;
return process(Who.accord, txn.ifBlock.get().conditional, lets(txn));
}
private Map<String, SelectResult> lets(Txn txn)
{
Map<String, SelectResult> lets = txn.lets.isEmpty() ? Map.of() : Maps.newHashMapWithExpectedSize(txn.lets.size());
for (Txn.Let let : txn.lets)
lets.put(let.symbol, getRowsAsByteBuffer(let.select));
return lets;
}
public void updateAndValidate(ByteBuffer[][] actual, Txn txn)
{
if (!shouldApply(txn)) return;
Map<String, SelectResult> lets = lets(txn);
txn.output.ifPresent(select -> validate(actual, select, lets));
List<Mutation> mutations = txn.ifBlock.isPresent()
? txn.ifBlock.get().mutations
: txn.mutations;
if (!mutations.isEmpty())
numMutations++; // bump here to make sure the last mutation doesn't have the same ts as the mutations here
long nowTs = numMutations;
for (var m : mutations)
{
if (m.timestampOrDefault(NO_TIMESTAMP) == NO_TIMESTAMP)
m = m.withTimestamp(nowTs);
updateInternal(m);
}
}
private void validate(ByteBuffer[][] actual, Select select, Map<String, SelectResult> lets)
{
if (select.source.isPresent())
{
// remove references
select = (Select) select.visit(new Visitor()
{
@Override
public Expression visit(Expression e)
{
if (!(e instanceof Reference)) return e;
var ref = (Reference) e;
return new Literal(extract(ref, lets), ref.type());
}
});
validate(actual, select);
}
else
{
ImmutableUniqueList<Symbol> columns = columns(select);
ByteBuffer[] values = new ByteBuffer[columns.size()];
int offset = 0;
for (var e : select.selections)
{
Object value;
if (e instanceof Reference)
value = extract((Reference) e, lets);
else
value = ExpressionEvaluator.eval(e);
Literal literal = new Literal(value, e.type());
values[offset++] = literal.valueEncoded();
}
validate(columns, actual, new ByteBuffer[][] {values});
}
}
public void update(Mutation mutation)
{
if (!shouldApply(mutation)) return;
@ -219,17 +322,23 @@ public class ASTSingleTableModel
{
if (!shouldApply(mutation))
{
if (mutation.isCas())
if (mutation.isCas() && validateCass.validate())
validateCasNotApplied(actual, mutation);
return;
}
if (mutation.isCas())
if (mutation.isCas() && validateCass.validate())
validate(CAS_APPLIED_COLUMNS, actual, CAS_SUCCESS_RESULT);
updateInternal(mutation);
}
private void validateCasNotApplied(ByteBuffer[][] actual, Mutation mutation)
{
if (validateCass == CasResponse.applied)
{
actual = new ByteBuffer[][] {new ByteBuffer[] {actual[0][0]}};
validate(CAS_APPLIED_COLUMNS, actual, CAS_REJECTION_RESULT);
return;
}
// see org.apache.cassandra.cql3.statements.ModificationStatement.buildCasFailureResultSet
var condition = mutation.casCondition().get();
var partition = partitions.get(referencePartition(mutation));
@ -641,7 +750,7 @@ public class ASTSingleTableModel
return Reference.of(rowSymbol, r);
}
});
return process(updatedCondition, lets);
return process(Who.cas, updatedCondition, lets);
}
public BytesPartitionState.Ref referencePartition(Mutation mutation)
@ -649,7 +758,9 @@ public class ASTSingleTableModel
return factory.createRef(pd(mutation));
}
private boolean process(Conditional condition, Map<String, SelectResult> lets)
private enum Who { cas, accord }
private boolean process(Who who, Conditional condition, Map<String, SelectResult> lets)
{
if (condition.getClass() == Conditional.Is.class)
{
@ -670,11 +781,28 @@ public class ASTSingleTableModel
ByteBuffer rhs = where.rhs instanceof ReferenceExpression
? (ByteBuffer) extract((ReferenceExpression) where.rhs, lets)
: eval(where.rhs);
// If anything is null avoid doing the test, but there is a special case where this returns true... both sides are null!
// This logic isn't consistent with other parts of the database and is local to CAS IF clause
// see ML@Inconsistent null handling between WHERE and IF clauses
if (lhs == null || rhs == null)
return lhs == rhs;
switch (who)
{
case cas:
{
// If anything is null avoid doing the test, but there is a special case where this returns true... both sides are null!
// This logic isn't consistent with other parts of the database and is local to CAS IF clause
// see ML@Inconsistent null handling between WHERE and IF clauses
if (lhs == null || rhs == null)
return lhs == rhs;
}
break;
case accord:
{
if (where.lhs.type().isNull(lhs))
return false;
if (where.rhs.type().isNull(rhs))
return false;
}
break;
default:
throw new UnsupportedOperationException(who.name());
}
return where.kind.test(where.lhs.type(), lhs, rhs);
}
else if (condition.getClass() == Conditional.And.class)
@ -682,7 +810,7 @@ public class ASTSingleTableModel
var conditions = condition.simplify();
for (var c : conditions)
{
if (!process(c, lets))
if (!process(who, c, lets))
return false;
}
return true;
@ -941,6 +1069,11 @@ public class ASTSingleTableModel
return partitions.get(ref);
}
public BytesPartitionState get(Clustering<ByteBuffer> key)
{
return get(factory.createRef(key));
}
public List<BytesPartitionState> getByToken(Token token)
{
NavigableSet<BytesPartitionState.Ref> keys = partitions.navigableKeySet();
@ -952,6 +1085,19 @@ public class ASTSingleTableModel
return matches.stream().map(partitions::get).collect(Collectors.toList());
}
public List<Clustering<ByteBuffer>> partitions(Select select)
{
if (select.where.isEmpty())
throw new IllegalArgumentException("Partition is full table scan, doesn't directly list partitions");
LookupContext ctx = context(select);
if (ctx.unmatchable)
throw new IllegalArgumentException("Select can not match anything");
if (ctx.eq.keySet().containsAll(factory.partitionColumns))
return keys(ctx.eq, factory.partitionColumns);
throw new IllegalArgumentException("Partition does not directly list any partitions");
}
public void validate(ByteBuffer[][] actual, Select select)
{
if (select.source.isEmpty())
@ -1141,7 +1287,7 @@ public class ASTSingleTableModel
private static Set<Row> toRow(ImmutableUniqueList<Symbol> columns, ByteBuffer[][] rows)
{
Set<Row> set = new HashSet<>();
Set<Row> set = new LinkedHashSet<>();
for (ByteBuffer[] row : rows)
set.add(new Row(columns, row));
return set;
@ -1199,17 +1345,23 @@ public class ASTSingleTableModel
}
return true;
}
@Override
public String toString()
{
return table(columns, rows);
}
}
private ImmutableUniqueList<Symbol> columns(Select select)
public ImmutableUniqueList<Symbol> columns(Select select)
{
if (select.selections.isEmpty()) return factory.selectionOrder;
var builder = ImmutableUniqueList.<Symbol>builder();
for (var e : select.selections)
{
if (!(e instanceof Symbol))
throw new UnsupportedOperationException("Only column selection is currently supported");
builder.add((Symbol) e);
if (e instanceof Symbol) builder.add((Symbol) e);
else if (e instanceof Reference) builder.add(new Symbol(e.name(), e.type())); //TODO (maintaince): should the columns be ReferenceExpression? this would allow foo['bar'] as a column, which could be fine?
else throw new UnsupportedOperationException("Only column selection or references are currently supported");
}
return builder.build();
}
@ -1649,7 +1801,7 @@ public class ASTSingleTableModel
ByteBuffer b = values[offset];
if (b == null) return "null";
if (ByteBufferUtil.EMPTY_BYTE_BUFFER.equals(b)) return "<empty>";
return symbol.type().asCQL3Type().toCQLLiteral(b);
return symbol.type().toCQLString(b);
}
public List<String> asCQL()

View File

@ -0,0 +1,76 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.harry.model;
import org.apache.cassandra.cql3.ast.Symbol;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.ImmutableUniqueList;
public class DetailedTableMetadata
{
public final TableMetadata metadata;
public final ImmutableUniqueList<Symbol> partitionColumns;
public final ImmutableUniqueList<Symbol> clusteringColumns;
public final ImmutableUniqueList<Symbol> primaryColumns;
public final ImmutableUniqueList<Symbol> staticColumns;
public final ImmutableUniqueList<Symbol> regularColumns;
public final ImmutableUniqueList<Symbol> selectionOrder, partitionAndStaticColumns, clusteringAndRegularColumns, regularAndStaticColumns;
public DetailedTableMetadata(TableMetadata metadata)
{
this.metadata = metadata;
ImmutableUniqueList.Builder<Symbol> symbolListBuilder = ImmutableUniqueList.builder();
for (ColumnMetadata pk : metadata.partitionKeyColumns())
symbolListBuilder.add(Symbol.from(pk));
partitionColumns = symbolListBuilder.buildAndClear();
for (ColumnMetadata pk : metadata.clusteringColumns())
symbolListBuilder.add(Symbol.from(pk));
clusteringColumns = symbolListBuilder.buildAndClear();
if (clusteringColumns.isEmpty()) primaryColumns = partitionColumns;
else
{
primaryColumns = symbolListBuilder.addAll(partitionColumns)
.addAll(clusteringColumns)
.buildAndClear();
}
metadata.staticColumns().selectOrderIterator().forEachRemaining(cm -> symbolListBuilder.add(Symbol.from(cm)));
staticColumns = symbolListBuilder.buildAndClear();
if (staticColumns.isEmpty()) partitionAndStaticColumns = partitionColumns;
else
{
partitionAndStaticColumns = symbolListBuilder.addAll(partitionColumns)
.addAll(staticColumns)
.buildAndClear();
}
metadata.regularColumns().selectOrderIterator().forEachRemaining(cm -> symbolListBuilder.add(Symbol.from(cm)));
regularColumns = symbolListBuilder.buildAndClear();
clusteringAndRegularColumns = symbolListBuilder.addAll(clusteringColumns)
.addAll(regularColumns)
.buildAndClear();
metadata.allColumnsInSelectOrder().forEachRemaining(cm -> symbolListBuilder.add(Symbol.from(cm)));
selectionOrder = symbolListBuilder.buildAndClear();
regularAndStaticColumns = symbolListBuilder.addAll(staticColumns).addAll(regularColumns).buildAndClear();
}
public Symbol find(String name)
{
return selectionOrder.stream().filter(s -> s.symbol.equals(name)).findAny().get();
}
}

View File

@ -125,7 +125,7 @@ public class BatchStatementBench
@Benchmark
public void bench()
{
bs.getMutations(ClientState.forInternalCalls(), bqo, false, nowInSec, nowInSec, queryStartTime);
bs.getMutations(ClientState.forInternalCalls(), bqo, false, nowInSec, nowInSec, queryStartTime, null);
}

View File

@ -0,0 +1,82 @@
/*
* 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.simulator;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.simulator.cluster.ClusterActions;
import org.apache.cassandra.simulator.systems.SimulatedSystems;
import org.apache.cassandra.utils.CloseableIterator;
public abstract class AbstractSimulation implements Simulation
{
public final SimulatedSystems simulated;
public final RunnableActionScheduler scheduler;
public final Cluster cluster;
public final ClusterActions clusterActions;
protected AbstractSimulation(SimulatedSystems simulated, RunnableActionScheduler scheduler, Cluster cluster)
{
this(simulated, scheduler, cluster, ClusterActions.simple(simulated, cluster));
}
protected AbstractSimulation(SimulatedSystems simulated, RunnableActionScheduler scheduler, Cluster cluster, ClusterActions.Options options)
{
this(simulated, scheduler, cluster, ClusterActions.simple(simulated, cluster, options));
}
protected AbstractSimulation(SimulatedSystems simulated, RunnableActionScheduler scheduler, Cluster cluster, ClusterActions clusterActions)
{
this.simulated = simulated;
this.scheduler = scheduler;
this.cluster = cluster;
this.clusterActions = clusterActions;
}
@Override
public void run()
{
try (CloseableIterator<?> iter = iterator())
{
while (iter.hasNext())
{
checkForErrors();
iter.next();
}
checkForErrors();
}
}
protected void checkForErrors()
{
if (simulated.failures.hasFailure())
{
AssertionError error = new AssertionError("Errors detected during simulation");
// don't care about the stack trace... the issue is the errors found and not what part of the scheduler we stopped
error.setStackTrace(new StackTraceElement[0]);
simulated.failures.get().forEach(error::addSuppressed);
throw error;
}
}
@Override
public void close() throws Exception
{
}
}

View File

@ -73,7 +73,34 @@ public class ActionSchedule implements CloseableIterator<Object>, LongConsumer
{
private static final Logger logger = LoggerFactory.getLogger(ActionList.class);
public enum Mode { TIME_LIMITED, STREAM_LIMITED, TIME_AND_STREAM_LIMITED, FINITE, UNLIMITED }
public enum Mode
{
/**
* Definition: Runs the simulation for a specific duration (specified by {@link Work#runForNanos})
* Behavior: After the time limit is reached, cancels both daemon tasks and stream actions
*/
TIME_LIMITED,
/**
* Definition: Runs until all finite streams are processed ({@link #activeFiniteStreamCount} reaches 0)
* Behavior: Once all finite streams complete, cancels daemon tasks
*/
STREAM_LIMITED,
/**
* Definition: Combines both time and stream limitations
* Behavior: Cancels daemon tasks if either all finite streams complete OR the time limit is reached
*/
TIME_AND_STREAM_LIMITED,
/**
* Definition: Processes a finite set of actions; does not allow stream actions
* Behavior: Processes a finite set of actions, if any action is added that has stream then the action will be rejected.
*/
FINITE,
/**
* Definition: Processes a finite set of actions, but allow daemon actions (unlike FINITE).
* Behavior: Similar to FINITE, but will handle daemon tasks in "waves", so that the scheduler can know when the "real" work is actually completed.
*/
UNLIMITED
}
public static class Work
{

View File

@ -141,6 +141,16 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
RunnableActionScheduler create(RandomSource random);
}
public interface FutureActionSchedulerFactory
{
FutureActionScheduler create(int nodeCount, SimulatedTime time, RandomSource random);
}
public interface PerVerbFutureActionSchedulersFactory
{
Map<Verb, FutureActionScheduler> create(int nodeCount, SimulatedTime time, RandomSource random);
}
@SuppressWarnings("UnusedReturnValue")
public static abstract class Builder<S extends Simulation>
{
@ -201,6 +211,8 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
protected SimulatedTime.Listener timeListener = (i1, i2) -> {};
protected LongConsumer onThreadLocalRandomCheck;
protected String transactionalMode = "off";
protected FutureActionSchedulerFactory futureActionSchedulerFactory;
protected PerVerbFutureActionSchedulersFactory perVerbFutureActionSchedulersFactory;
public Builder<S> failures(Failures failures)
{
@ -506,8 +518,16 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
return this;
}
public Builder<S> futureActionScheduler(FutureActionSchedulerFactory factory)
{
futureActionSchedulerFactory = factory;
return this;
}
public FutureActionScheduler futureActionScheduler(int nodeCount, SimulatedTime time, RandomSource random)
{
if (futureActionSchedulerFactory != null)
return futureActionSchedulerFactory.create(nodeCount, time, random);
KindOfSequence kind = Choices.random(random, KindOfSequence.values())
.choose(random);
return new SimulatedFutureActionScheduler(kind, nodeCount, random, time,
@ -517,8 +537,16 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
new SchedulerConfig(schedulerDelayChance, schedulerDelayNanos, schedulerLongDelayNanos));
}
public Builder<S> perVerbFutureActionSchedulers(PerVerbFutureActionSchedulersFactory factory)
{
perVerbFutureActionSchedulersFactory = factory;
return this;
}
public Map<Verb, FutureActionScheduler> perVerbFutureActionSchedulers(int nodeCount, SimulatedTime time, RandomSource random, FutureActionScheduler defaultScheduler)
{
if (perVerbFutureActionSchedulersFactory != null)
return perVerbFutureActionSchedulersFactory.create(nodeCount, time, random);
return Collections.emptyMap();
}
@ -716,7 +744,7 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
{
int numInDc = (numOfNodes / numOfDcs) + (numOfNodes % numOfDcs > i ? 1 : 0);
numInDcs[i] = numInDc;
minRf[i] = 3;
minRf[i] = min(numInDc, 3);
maxRf[i] = min(numInDc, 9);
initialRf[i] = random.uniform(minRf[i], 1 + maxRf[i]);
nc += numInDc;

View File

@ -21,6 +21,8 @@ package org.apache.cassandra.simulator.cluster;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.EnumMap;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@ -34,6 +36,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.Config.PaxosVariant;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.marshal.Int32Type;
@ -42,18 +45,23 @@ import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.IInstance;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.IIsolatedExecutor;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.locator.ReplicaLayout;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.simulator.Action;
import org.apache.cassandra.simulator.ActionList;
import org.apache.cassandra.simulator.Actions;
import org.apache.cassandra.simulator.Actions.StrictAction;
import org.apache.cassandra.simulator.Debug;
import org.apache.cassandra.simulator.RandomSource.Choices;
import org.apache.cassandra.simulator.systems.InterceptedExecution;
import org.apache.cassandra.simulator.systems.InterceptingExecutor;
import org.apache.cassandra.simulator.systems.NonInterceptible;
import org.apache.cassandra.simulator.systems.SimulatedActionTask;
import org.apache.cassandra.simulator.systems.SimulatedSystems;
import org.apache.cassandra.simulator.utils.KindOfSequence;
import org.apache.cassandra.tcm.ClusterMetadata;
@ -186,6 +194,19 @@ public class ClusterActions extends SimulatedSystems
this.debug = debug;
}
public static ClusterActions simple(SimulatedSystems simulated, Cluster cluster)
{
return simple(simulated, cluster, Options.noActions(cluster.size()));
}
public static ClusterActions simple(SimulatedSystems simulated, Cluster cluster, Options options)
{
return new ClusterActions(simulated, cluster, options,
new ClusterActionListener.NoOpListener(),
new Debug(new EnumMap<>(Debug.Info.class),
new int[0]));
}
public static class InitialConfiguration
{
public static final int[] EMPTY = {};
@ -268,6 +289,84 @@ public class ClusterActions extends SimulatedSystems
return transitivelyReliable("Reset Gossip", i, () -> Gossiper.runInGossipStageBlocking(Gossiper.instance::unsafeSetEnabled));
}
public Action reconfigureCMS(int node, int rf)
{
return reconfigureCMS(node, rf, false);
}
public Action reconfigureCMS(int nodeId, int rf, boolean inEachDc)
{
String caption = String.format("Reconfigure CMS rf=%d, inEachDc=%s", rf, inEachDc);
IInvokableInstance node = cluster.get(nodeId);
return new SimulatedActionTask(caption, Action.Modifiers.RELIABLE_NO_TIMEOUTS, Action.Modifiers.RELIABLE_NO_TIMEOUTS, null, this,
new InterceptedExecution.InterceptedRunnableExecution((InterceptingExecutor) node.executor(),
node.transfer((IIsolatedExecutor.SerializableRunnable) () -> {
ReplicationParams params;
if (inEachDc)
{
Map<String, Integer> rfs = new HashMap<>();
for (String dc : ClusterMetadata.current().directory.knownDatacenters())
{
rfs.put(dc, rf);
}
params = ReplicationParams.ntsMeta(rfs);
}
else
{
params = ReplicationParams.simpleMeta(rf, ClusterMetadata.current().directory.knownDatacenters());
}
ClusterMetadataService.instance().reconfigureCMS(params);
})));
}
public Action flush(String keyspace, String... tableNames)
{
return new Actions.ReliableAction("Flush on all nodes", () -> {
List<Action> actions = new ArrayList<>(cluster.size());
for (int i = 0; i < cluster.size(); i++)
actions.add(flush(i + 1, keyspace, tableNames));
return ActionList.of(actions).setStrictlySequential();
}, true);
}
public Action flush(int nodeId, String keyspace, String... tableNames)
{
String caption = String.format("Flush %s: %s", keyspace, Arrays.toString(tableNames));
IInvokableInstance node = cluster.get(nodeId);
return new SimulatedActionTask(caption, Action.Modifiers.RELIABLE_NO_TIMEOUTS, Action.Modifiers.RELIABLE_NO_TIMEOUTS, null, this,
new InterceptedExecution.InterceptedRunnableExecution((InterceptingExecutor) node.executor(),
node.transfer((IIsolatedExecutor.SerializableRunnable) () -> {
for (ColumnFamilyStore store : StorageService.instance.getValidColumnFamilies(false, false, keyspace, tableNames))
{
store.forceFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
}
})));
}
public Action compact(String keyspace, String... tableNames)
{
return new Actions.ReliableAction("Compact on all nodes", () -> {
List<Action> actions = new ArrayList<>(cluster.size());
for (int i = 0; i < cluster.size(); i++)
actions.add(compact(i + 1, keyspace, tableNames));
return ActionList.of(actions).setStrictlySequential();
}, true);
}
public Action compact(int nodeId, String keyspace, String... tableNames)
{
String caption = String.format("Compact %s: %s", keyspace, Arrays.toString(tableNames));
IInvokableInstance node = cluster.get(nodeId);
return new SimulatedActionTask(caption, Action.Modifiers.RELIABLE_NO_TIMEOUTS, Action.Modifiers.RELIABLE_NO_TIMEOUTS, null, this,
new InterceptedExecution.InterceptedRunnableExecution((InterceptingExecutor) node.executor(),
node.transfer((IIsolatedExecutor.SerializableRunnable) () -> {
for (ColumnFamilyStore store : StorageService.instance.getValidColumnFamilies(false, false, keyspace, tableNames))
{
store.submitMajorCompaction(false, -1);
}
})));
}
@SuppressWarnings("unchecked")
void validateReplicasForKeys(IInvokableInstance on, String keyspace, String table, Topology topology)
{

View File

@ -58,7 +58,9 @@ public interface InterceptibleThreadFactory extends ThreadFactory
@Override
protected synchronized InterceptibleThread newThread(ThreadGroup threadGroup, Runnable runnable, String name)
{
InterceptibleThread thread = new InterceptibleThread(threadGroup, runnable, name, extraToStringInfo, onTermination, parent.interceptorOfGlobalMethods, time);
// Can not use NamedThreadFactory.globalPrefix() as this method runs in the App class loader and not the Instance class loader; the ThreadGroup's name can act as a proxy for this.
String threadName = threadGroup.getName() + '_' + name;
InterceptibleThread thread = new InterceptibleThread(threadGroup, runnable, threadName, extraToStringInfo, onTermination, parent.interceptorOfGlobalMethods, time);
if (parent.isClosed)
thread.trapInterrupts(false);
return setupThread(thread);
@ -79,7 +81,7 @@ public interface InterceptibleThreadFactory extends ThreadFactory
@Override
protected Thread newThread(ThreadGroup threadGroup, Runnable runnable, String name)
{
return super.newThread(threadGroup, () -> { try { runnable.run(); } finally { onTermination.run();} }, name);
return super.newThread(threadGroup, () -> { try { runnable.run(); } finally { onTermination.run();} }, threadGroup.getName() + '_' + name);
}
}

View File

@ -274,7 +274,7 @@ public class SimulatedTime
nextDrift = nanosDriftSupplier.get(random);
from = global;
to = global + Math.max(baseDrift, nextDrift);
diffPerGlobal = (nextDrift - baseDrift) / (double)(to - from);
diffPerGlobal = to == from ? 1 : (nextDrift - baseDrift) / (double)(to - from);
listener.accept("SetNextDrift", nextDrift);
}

View File

@ -28,6 +28,7 @@ import java.util.function.Consumer;
import java.util.function.Supplier;
import com.google.common.util.concurrent.Uninterruptibles;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
@ -123,6 +124,7 @@ import static org.apache.cassandra.simulator.cluster.ClusterActions.Options.noAc
--add-opens java.desktop/com.sun.beans.introspect=ALL-UNNAMED
*/
@Ignore("Something is currently wrong with SimulationTestBase and multi node tests. CASSANDRA-20744 exposed that simulator was swallowing errors which then causes this test to fail 100% of the time; until the root cause is fixed it doesn't make sense to run this test.")
public class EpochStressTest extends SimulationTestBase
{
@Test

View File

@ -25,6 +25,7 @@ import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.IntSupplier;
import java.util.function.LongConsumer;
import java.util.function.LongSupplier;
import java.util.function.Predicate;
import com.google.common.collect.Iterators;
@ -35,11 +36,13 @@ 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.IInvokableInstance;
import org.apache.cassandra.distributed.api.IInstanceConfig;
import org.apache.cassandra.distributed.api.IIsolatedExecutor;
import org.apache.cassandra.distributed.api.IMessage;
import org.apache.cassandra.distributed.impl.AbstractCluster;
import org.apache.cassandra.distributed.impl.IsolatedExecutor;
import org.apache.cassandra.distributed.shared.InstanceClassLoader;
import org.apache.cassandra.simulator.AbstractSimulation;
import org.apache.cassandra.simulator.Action;
import org.apache.cassandra.simulator.ActionList;
import org.apache.cassandra.simulator.ActionPlan;
@ -51,9 +54,11 @@ import org.apache.cassandra.simulator.FutureActionScheduler;
import org.apache.cassandra.simulator.RandomSource;
import org.apache.cassandra.simulator.RunnableActionScheduler;
import org.apache.cassandra.simulator.Simulation;
import org.apache.cassandra.simulator.SimulationException;
import org.apache.cassandra.simulator.SimulationRunner;
import org.apache.cassandra.simulator.asm.InterceptClasses;
import org.apache.cassandra.simulator.asm.NemesisFieldSelectors;
import org.apache.cassandra.simulator.cluster.ClusterActions;
import org.apache.cassandra.simulator.systems.Failures;
import org.apache.cassandra.simulator.systems.InterceptedWait;
import org.apache.cassandra.simulator.systems.InterceptibleThread;
@ -80,6 +85,7 @@ import static org.apache.cassandra.simulator.ActionSchedule.Mode.UNLIMITED;
import static org.apache.cassandra.simulator.ClusterSimulation.ISOLATE;
import static org.apache.cassandra.simulator.ClusterSimulation.SHARE;
import static org.apache.cassandra.simulator.SimulatorUtils.failWithOOM;
import static org.apache.cassandra.simulator.cluster.ClusterActions.InitialConfiguration.initializeAll;
import static org.apache.cassandra.simulator.utils.KindOfSequence.UNIFORM;
import static org.apache.cassandra.utils.Shared.Scope.ANY;
import static org.apache.cassandra.utils.Shared.Scope.SIMULATION;
@ -98,17 +104,22 @@ public class SimulationTestBase
// Don't use loggers before invoking simulator it messes up initialization order
// private static final Logger logger = LoggerFactory.getLogger(Logger.class);
static abstract class DTestClusterSimulation implements Simulation
{
protected final SimulatedSystems simulated;
protected final RunnableActionScheduler scheduler;
protected final Cluster cluster;
public DTestClusterSimulation(SimulatedSystems simulated, RunnableActionScheduler scheduler, Cluster cluster)
static abstract class SimpleSimulation extends AbstractSimulation
{
protected SimpleSimulation(SimulatedSystems simulated, RunnableActionScheduler scheduler, Cluster cluster)
{
this.simulated = simulated;
this.scheduler = scheduler;
this.cluster = cluster;
super(simulated, scheduler, cluster);
}
protected SimpleSimulation(SimulatedSystems simulated, RunnableActionScheduler scheduler, Cluster cluster, ClusterActions.Options options)
{
super(simulated, scheduler, cluster, options);
}
protected SimpleSimulation(SimulatedSystems simulated, RunnableActionScheduler scheduler, Cluster cluster, ClusterActions clusterActions)
{
super(simulated, scheduler, cluster, clusterActions);
}
public Action executeQuery(int node, String query, ConsistencyLevel cl, Object... bindings)
@ -132,56 +143,79 @@ public class SimulationTestBase
null);
}
protected abstract ActionList initialize();
protected abstract ActionList teardown();
protected ActionList initialize()
{
return ActionList.of(clusterActions.initializeCluster(initializeAll(cluster.size())));
}
protected ActionList teardown()
{
return ActionList.of();
}
protected abstract ActionList execute();
protected ActionSchedule.Mode mode()
{
return TIME_LIMITED;
}
protected long runForNanos()
{
return MINUTES.toNanos(10);
}
@Override
public CloseableIterator<?> iterator()
{
return ActionPlan.setUpTearDown(ActionList.of(initialize()),
ActionList.of(teardown()))
return ActionPlan.setUpTearDown(initialize(),
teardown())
.encapsulate(ActionPlan.interleave(Collections.singletonList(execute())))
.iterator(TIME_LIMITED, MINUTES.toNanos(10), () -> 0L, simulated.time, scheduler, simulated.futureScheduler);
}
public void run()
{
try (CloseableIterator<?> iter = iterator())
{
while (iter.hasNext())
iter.next();
}
}
public void close() throws Exception
{
.iterator(mode(), runForNanos(), () -> 0L, simulated.time, scheduler, simulated.futureScheduler);
}
}
static class DTestClusterSimulationBuilder extends ClusterSimulation.Builder<DTestClusterSimulation>
static abstract class BasicSimulationBuilder<S extends Simulation> extends ClusterSimulation.Builder<S>
{
protected final Function<DTestClusterSimulation, ActionList> init;
protected final Function<DTestClusterSimulation, ActionList> test;
protected final Function<DTestClusterSimulation, ActionList> teardown;
abstract S create(SimulatedSystems simulated, RunnableActionScheduler scheduler, Cluster cluster, ClusterActions.Options options);
DTestClusterSimulationBuilder(Function<DTestClusterSimulation, ActionList> init,
Function<DTestClusterSimulation, ActionList> test,
Function<DTestClusterSimulation, ActionList> teardown)
protected void updateConfig(IInstanceConfig config)
{
}
public ClusterSimulation<S> create(long seed) throws IOException
{
RandomSource random = new RandomSource.Default();
random.reset(seed);
return new ClusterSimulation<>(random, seed, 1, this,
this::updateConfig,
this::create);
}
}
static class DTestClusterSimulationBuilder extends ClusterSimulation.Builder<SimpleSimulation>
{
protected final Function<SimpleSimulation, ActionList> init;
protected final Function<SimpleSimulation, ActionList> test;
protected final Function<SimpleSimulation, ActionList> teardown;
DTestClusterSimulationBuilder(Function<SimpleSimulation, ActionList> init,
Function<SimpleSimulation, ActionList> test,
Function<SimpleSimulation, ActionList> teardown)
{
this.init = init;
this.test = test;
this.teardown = teardown;
}
public ClusterSimulation<DTestClusterSimulation> create(long seed) throws IOException
public ClusterSimulation<SimpleSimulation> create(long seed) throws IOException
{
RandomSource random = new RandomSource.Default();
random.reset(seed);
return new ClusterSimulation<>(random, seed, 1, this,
(c) -> {},
(simulated, scheduler, cluster, options) -> new DTestClusterSimulation(simulated, scheduler, cluster)
(simulated, scheduler, cluster, options) -> new SimpleSimulation(simulated, scheduler, cluster)
{
protected ActionList initialize()
{
@ -201,10 +235,10 @@ public class SimulationTestBase
}
}
public static void simulate(Function<DTestClusterSimulation, ActionList> init,
Function<DTestClusterSimulation, ActionList> test,
Function<DTestClusterSimulation, ActionList> teardown,
Consumer<ClusterSimulation.Builder<DTestClusterSimulation>> configure) throws IOException
public static void simulate(Function<SimpleSimulation, ActionList> init,
Function<SimpleSimulation, ActionList> test,
Function<SimpleSimulation, ActionList> teardown,
Consumer<ClusterSimulation.Builder<SimpleSimulation>> configure) throws IOException
{
simulate(new DTestClusterSimulationBuilder(init, test, teardown),
configure);
@ -212,9 +246,49 @@ public class SimulationTestBase
public static <T extends Simulation> void simulate(ClusterSimulation.Builder<T> factory,
Consumer<ClusterSimulation.Builder<T>> configure) throws IOException
{
simulate(System::currentTimeMillis, factory, configure);
}
public static <T extends Simulation> void simulate(long seed, ClusterSimulation.Builder<T> factory) throws IOException
{
simulate(() -> seed, factory, i ->{});
}
public static <T extends Simulation> void simulate(ClusterSimulation.SimulationFactory<T> factory) throws IOException
{
simulate(System.currentTimeMillis(), factory);
}
public static <T extends Simulation> void simulate(long seed, ClusterSimulation.SimulationFactory<T> factory) throws IOException
{
simulate(seed, factory, b -> {});
}
public static <T extends Simulation> void simulate(ClusterSimulation.SimulationFactory<T> factory, Consumer<ClusterSimulation.Builder<T>> configure) throws IOException
{
simulate(System.currentTimeMillis(), factory, configure);
}
public static <T extends Simulation> void simulate(long seed, ClusterSimulation.SimulationFactory<T> factory, Consumer<ClusterSimulation.Builder<T>> configure) throws IOException
{
BasicSimulationBuilder builder = new BasicSimulationBuilder()
{
@Override
Simulation create(SimulatedSystems simulated, RunnableActionScheduler scheduler, Cluster cluster, ClusterActions.Options options)
{
return factory.create(simulated, scheduler, cluster, options);
}
};
simulate(() -> seed, builder, configure);
}
public static <T extends Simulation> void simulate(LongSupplier seedGen,
ClusterSimulation.Builder<T> factory,
Consumer<ClusterSimulation.Builder<T>> configure) throws IOException
{
SimulationRunner.beforeAll();
long seed = System.currentTimeMillis();
long seed = seedGen.getAsLong();
// Development seed:
//long seed = 1687184561194L;
System.out.printf("Simulation seed: %dL%n", seed);
@ -227,10 +301,15 @@ public class SimulationTestBase
}
catch (Throwable t)
{
throw new AssertionError(String.format("Failed on seed %s", Long.toHexString(seed)),
t);
throw new SimulationException(seed, t);
}
}
catch (Throwable t)
{
if (t instanceof SimulationException)
throw t;
throw new SimulationException(seed, t);
}
}
public static void simulate(IIsolatedExecutor.SerializableRunnable run,

View File

@ -0,0 +1,166 @@
/*
* 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.simulator.test;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.junit.Ignore;
import org.junit.Test;
import accord.utils.DefaultRandom;
import accord.utils.Gen;
import accord.utils.Gens;
import accord.utils.RandomSource;
import accord.utils.SeedProvider;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.simulator.AlwaysDeliverNetworkScheduler;
import org.apache.cassandra.simulator.ClusterSimulation;
import org.apache.cassandra.simulator.FutureActionScheduler;
import static java.util.concurrent.TimeUnit.SECONDS;
/**
* In order to run these tests in your IDE, you need to first build a simulator jara
*
* ant simulator-jars
*
* And then run your test using the following settings (omit add-* if you are running on jdk8):
*
-Dstorage-config=$MODULE_DIR$/test/conf
-Djava.awt.headless=true
-javaagent:$MODULE_DIR$/lib/jamm-0.4.0.jar
-ea
-Dcassandra.debugrefcount=true
-Xss384k
-XX:SoftRefLRUPolicyMSPerMB=0
-XX:ActiveProcessorCount=2
-XX:HeapDumpPath=build/test
-Dcassandra.test.driver.connection_timeout_ms=10000
-Dcassandra.test.driver.read_timeout_ms=24000
-Dcassandra.memtable_row_overhead_computation_step=100
-Dcassandra.test.use_prepared=true
-Dcassandra.test.sstableformatdevelopment=true
-Djava.security.egd=file:/dev/urandom
-Dcassandra.testtag=.jdk11
-Dcassandra.keepBriefBrief=true
-Dcassandra.allow_simplestrategy=true
-Dcassandra.strict.runtime.checks=true
-Dcassandra.reads.thresholds.coordinator.defensive_checks_enabled=true
-Dcassandra.test.flush_local_schema_changes=false
-Dcassandra.test.messagingService.nonGracefulShutdown=true
-Dcassandra.use_nix_recursive_delete=true
-Dcie-cassandra.disable_schema_drop_log=true
-Dlogback.configurationFile=file://$MODULE_DIR$/test/conf/logback-simulator.xml
-Dcassandra.ring_delay_ms=10000
-Dcassandra.tolerate_sstable_size=true
-Dcassandra.skip_sync=true
-Dcassandra.debugrefcount=false
-Dcassandra.test.simulator.determinismcheck=strict
-Dcassandra.test.simulator.print_asm=none
-javaagent:$MODULE_DIR$/build/test/lib/jars/simulator-asm.jar
-Xbootclasspath/a:$MODULE_DIR$/build/test/lib/jars/simulator-bootstrap.jar
-XX:ActiveProcessorCount=4
-XX:-TieredCompilation
-XX:-BackgroundCompilation
-XX:CICompilerCount=1
-XX:Tier4CompileThreshold=1000
-XX:ReservedCodeCacheSize=256M
-Xmx16G
-Xmx4G
--add-exports java.base/jdk.internal.misc=ALL-UNNAMED
--add-exports java.base/jdk.internal.ref=ALL-UNNAMED
--add-exports java.base/sun.nio.ch=ALL-UNNAMED
--add-exports java.management.rmi/com.sun.jmx.remote.internal.rmi=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 java.rmi/sun.rmi.registry=ALL-UNNAMED
--add-opens java.base/java.lang.module=ALL-UNNAMED
--add-opens java.base/java.net=ALL-UNNAMED
--add-opens java.base/jdk.internal.loader=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.math=ALL-UNNAMED
--add-opens java.base/jdk.internal.module=ALL-UNNAMED
--add-opens java.base/jdk.internal.util.jar=ALL-UNNAMED
--add-opens jdk.management/com.sun.management.internal=ALL-UNNAMED
--add-opens jdk.management.jfr/jdk.management.jfr=ALL-UNNAMED
--add-opens java.desktop/com.sun.beans.introspect=ALL-UNNAMED
*/
public class SingleNodeSingleTableASTTest extends SimulationTestBase
{
private static final Gen<Gen.IntGen> VERB_DELAY_DISTRIBUTION_MS = Gens.ints().mixedDistribution(10, 1000);
/**
* Simulator requires that loggers are not setup before simulator starts, but the generators below touch {@link accord.utils.Invariants} which
* allocates a logger which leads to the following errors
*
* <pre>{@code Caused by: java.lang.NoClassDefFoundError: Could not initialize class org.apache.cassandra.simulator.logging.RunStartDefiner}</pre>
*/
private static class LazyInit
{
private static final Gen.IntGen THREAD_GEN = Gens.pickInt(10, 100, 1_000);
}
@Test
@Ignore
public void normal() throws IOException
{
// testOne(SimulationRunner.parseHex("0x2fd91c2a2be59d7d"), SingleTableASTSimulation::new);
testOne(SeedProvider.instance.nextSeed(), SingleTableASTSimulation::new);
}
@Test
public void accordFull() throws IOException
{
// testOne(SimulationRunner.parseHex("0xd72a3d79134c4dbb"), SingleTableASTSimulation.FullAccordSingleTableASTSimulation::new);
testOne(SeedProvider.instance.nextSeed(), SingleTableASTSimulation.FullAccordSingleTableASTSimulation::new);
}
@Test
@Ignore
public void accordMixedReads() throws IOException
{
// testOne(SimulationRunner.parseHex("0x2fd91c2a2be59d7d"), SingleTableASTSimulation.MixedReadsAccordSingleTableASTSimulation::new);
testOne(SeedProvider.instance.nextSeed(), SingleTableASTSimulation.MixedReadsAccordSingleTableASTSimulation::new);
}
private void testOne(long seed, ClusterSimulation.SimulationFactory<SingleTableASTSimulation> factory) throws IOException
{
RandomSource rs = new DefaultRandom(seed);
Gen.IntGen delayMillis = VERB_DELAY_DISTRIBUTION_MS.next(rs);
simulate(seed, factory, b ->
b.futureActionScheduler((i1, time, i2) -> new AlwaysDeliverNetworkScheduler(time))
.perVerbFutureActionSchedulers((i1, time, i2) -> {
Map<Verb, FutureActionScheduler> map = new HashMap<>();
for (Verb verb : Verb.values())
map.put(verb, new AlwaysDeliverNetworkScheduler(time, TimeUnit.MILLISECONDS.toNanos(delayMillis.nextInt(rs))));
return map;
})
.writeTimeoutNanos(SECONDS.toNanos(120))
.readTimeoutNanos(SECONDS.toNanos(120))
.requestTimeoutNanos(SECONDS.toNanos(120))
.threadCount(LazyInit.THREAD_GEN.nextInt(rs))
.nodes(1, 1)
.dcs(1, 1));
}
}

View File

@ -0,0 +1,408 @@
/*
* 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.simulator.test;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.NavigableSet;
import java.util.function.Consumer;
import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.utils.DefaultRandom;
import accord.utils.Gen;
import accord.utils.Gens;
import accord.utils.RandomSource;
import org.apache.cassandra.cql3.ast.CQLFormatter;
import org.apache.cassandra.cql3.ast.Mutation;
import org.apache.cassandra.cql3.ast.Select;
import org.apache.cassandra.cql3.ast.StandardVisitors;
import org.apache.cassandra.cql3.ast.Statement;
import org.apache.cassandra.cql3.ast.Symbol;
import org.apache.cassandra.cql3.ast.Txn;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.marshal.AsciiType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.IIsolatedExecutor;
import org.apache.cassandra.distributed.impl.Query;
import org.apache.cassandra.distributed.impl.RowUtil;
import org.apache.cassandra.harry.model.ASTSingleTableModel;
import org.apache.cassandra.harry.model.BytesPartitionState;
import org.apache.cassandra.harry.util.StringUtils;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
import org.apache.cassandra.simulator.AbstractSimulation;
import org.apache.cassandra.simulator.Action;
import org.apache.cassandra.simulator.ActionList;
import org.apache.cassandra.simulator.ActionSchedule;
import org.apache.cassandra.simulator.Actions;
import org.apache.cassandra.simulator.RunnableActionScheduler;
import org.apache.cassandra.simulator.cluster.ClusterActions;
import org.apache.cassandra.simulator.systems.SimulatedActionCallable;
import org.apache.cassandra.simulator.systems.SimulatedSystems;
import org.apache.cassandra.utils.ASTGenerators;
import org.apache.cassandra.utils.AbstractTypeGenerators;
import org.apache.cassandra.utils.CassandraGenerators;
import org.apache.cassandra.utils.FastByteOperations;
import org.apache.cassandra.utils.Generators;
import org.quicktheories.generators.SourceDSL;
import static org.apache.cassandra.simulator.cluster.ClusterActions.InitialConfiguration.initializeAll;
import static org.apache.cassandra.utils.AbstractTypeGenerators.overridePrimitiveTypeSupport;
import static org.apache.cassandra.utils.AbstractTypeGenerators.stringComparator;
import static org.apache.cassandra.utils.Generators.toGen;
public class SingleTableASTSimulation extends SimulationTestBase.SimpleSimulation
{
private static final int MAX_STEPS = 5000;
static
{
// limit text/bytes so they are not too big; mostly for debugging than anything
overridePrimitiveTypeSupport(AsciiType.instance, AbstractTypeGenerators.TypeSupport.of(AsciiType.instance, SourceDSL.strings().ascii().ofLengthBetween(1, 10), stringComparator(AsciiType.instance)));
overridePrimitiveTypeSupport(UTF8Type.instance, AbstractTypeGenerators.TypeSupport.of(UTF8Type.instance, Generators.utf8(1, 10), stringComparator(UTF8Type.instance)));
overridePrimitiveTypeSupport(BytesType.instance, AbstractTypeGenerators.TypeSupport.of(BytesType.instance, Generators.bytes(1, 10), FastByteOperations::compareUnsigned));
}
private final RandomSource rs;
private ASTRunner runner;
protected SingleTableASTSimulation(SimulatedSystems simulated, RunnableActionScheduler scheduler, Cluster cluster, ClusterActions.Options options)
{
super(simulated, scheduler, cluster, options);
this.rs = new DefaultRandom(simulated.random.uniform(Long.MIN_VALUE, Long.MAX_VALUE));
cluster.stream().forEach((IInvokableInstance i) -> simulated.failureDetector.markUp(i.config().broadcastAddress()));
}
@Override
protected ActionSchedule.Mode mode()
{
return ActionSchedule.Mode.STREAM_LIMITED;
}
@Override
protected ActionList initialize()
{
List<Action> actions = new ArrayList<>();
actions.add(clusterActions.initializeCluster(initializeAll(cluster.size())));
int[] dcSizes = new int[clusterActions.snitch.dcCount()];
for (int nodeId = 1; nodeId <= cluster.size(); nodeId++)
dcSizes[clusterActions.snitch.dcOf(nodeId)]++;
actions.addAll(setupTable(dcSizes));
return ActionList.of(actions);
}
protected List<Action> setupTable(int[] dcSizes)
{
List<Action> actions = new ArrayList<>();
StringBuilder createKeyspace = new StringBuilder("CREATE KEYSPACE ks WITH replication = {'class': 'NetworkTopologyStrategy'"); // when the simulation starts the RF gets populated
for (int i = 0; i < dcSizes.length; i++)
{
String name = clusterActions.snitch.nameOfDc(i);
createKeyspace.append(", '").append(name).append("': ").append(Math.min(3, dcSizes[i]));
}
createKeyspace.append("};");
actions.add(clusterActions.schemaChange(1, createKeyspace.toString()));
TableMetadata metadata = defineTable(rs, "ks");
CassandraGenerators.visitUDTs(metadata, udt -> actions.add(clusterActions.schemaChange(1, udt.toCqlString(false, false, false))));
actions.add(clusterActions.schemaChange(1, metadata.toCqlString(false, false, false)));
this.runner = new ASTRunner(metadata, rs, MAX_STEPS, this);
return actions;
}
protected AbstractTypeGenerators.TypeGenBuilder supportedTypes()
{
return AbstractTypeGenerators.withoutUnsafeEquality()
.withTypeKinds(AbstractTypeGenerators.TypeKind.PRIMITIVE);
}
protected TableMetadata defineTable(RandomSource rs, String ks)
{
TableMetadata tbl = toGen(new CassandraGenerators.TableMetadataBuilder()
.withTableKinds(TableMetadata.Kind.REGULAR)
.withKnownMemtables()
.withKeyspaceName(ks).withTableName("tbl")
.withSimpleColumnNames()
.withDefaultTypeGen(supportedTypes())
.withPartitioner(Murmur3Partitioner.instance)
.build())
.next(rs);
return tbl.unbuild().params(tbl.params.unbuild().readRepair(ReadRepairStrategy.NONE).build()).build();
}
@Override
protected ActionList execute()
{
return ActionList.of(test());
}
protected Action test()
{
return Actions.stream(1, runner::next);
}
public static class FullAccordSingleTableASTSimulation extends SingleTableASTSimulation
{
public FullAccordSingleTableASTSimulation(SimulatedSystems simulated, RunnableActionScheduler scheduler, Cluster cluster, ClusterActions.Options options)
{
super(simulated, scheduler, cluster, options);
}
@Override
protected TableMetadata defineTable(RandomSource rs, String ks)
{
TableMetadata metadata = super.defineTable(rs, ks);
return metadata.unbuild()
.params(metadata.params.unbuild()
.transactionalMode(TransactionalMode.full)
.build())
.build();
}
}
public static class MixedReadsAccordSingleTableASTSimulation extends SingleTableASTSimulation
{
public MixedReadsAccordSingleTableASTSimulation(SimulatedSystems simulated, RunnableActionScheduler scheduler, Cluster cluster, ClusterActions.Options options)
{
super(simulated, scheduler, cluster, options);
}
@Override
protected TableMetadata defineTable(RandomSource rs, String ks)
{
TableMetadata metadata = super.defineTable(rs, ks);
return metadata.unbuild()
.params(metadata.params.unbuild()
.transactionalMode(TransactionalMode.mixed_reads)
.build())
.build();
}
}
public static class ASTRunner
{
private static final Logger logger = LoggerFactory.getLogger(ASTRunner.class);
private final TableMetadata metadata;
private final ASTSingleTableModel model;
private final RandomSource rs;
private final int maxSteps;
private final Gen<Action> commands;
private final AbstractSimulation simulation;
private int step;
public ASTRunner(TableMetadata metadata, RandomSource rs, int maxSteps, AbstractSimulation simulation)
{
this.metadata = metadata;
this.model = new ASTSingleTableModel(metadata);
this.rs = rs;
this.maxSteps = maxSteps;
this.simulation = simulation;
Gen.IntGen nodeGen = r -> r.nextInt(0, simulation.cluster.size()) + 1;
List<LinkedHashMap<Symbol, Object>> uniquePartitions = Gens.lists(toGen(ASTGenerators.columnValues(model.factory.partitionColumns)))
.uniqueBestEffort()
.ofSize(rs.nextInt(1, 20))
.next(rs);
Gen<Action> mutationGen = toGen(ASTGenerators.mutationBuilder(rs, model, uniquePartitions, i -> null).disallowEmpty().build())
.map(mutation -> query(mutation));
Gen<Action> selectPartitionGen = Gens.pick(uniquePartitions)
.map(partition -> query(select(partition).build()));
Gen<Action> selectRowGen = Gens.pick(uniquePartitions).map(this::selectRow);
Gen<Action> txnGen = Generators.toGen(new ASTGenerators.ModelBasedTxnGenBuilder(rs, model, i -> null, SourceDSL.arbitrary().pick(uniquePartitions)).disallowEmpty().build())
.map(txn -> query(txn));
Gens.OneOfBuilder<Action> commandsBuilder = Gens.<Action>oneOf()
.add(mutationGen)
.add(selectPartitionGen)
.add(i -> query(Select.builder(metadata).build()))
.add(r -> {
int nodeId = nodeGen.nextInt(r);
logger.warn("[step={}] Scheduling flush on node{}", step, nodeId);
return simulation.clusterActions.flush(nodeId, metadata.keyspace, metadata.name);
})
.add(r -> {
int nodeId = nodeGen.nextInt(r);
logger.warn("[step={}] Scheduling compact on node{}", step, nodeId);
return simulation.clusterActions.compact(nodeId, metadata.keyspace, metadata.name);
});
if (!model.factory.clusteringColumns.isEmpty())
commandsBuilder.add(selectRowGen);
if (metadata.params.transactionalMode.accordIsEnabled)
commandsBuilder.add(txnGen);
this.commands = commandsBuilder.buildWithDynamicWeights().next(rs);
}
public Action next()
{
if (step + 1 >= maxSteps)
{
logger.warn("Max steps reached, exiting...");
return null;
}
step++;
Action next = commands.next(rs);
// empty actions implies that the model state doesn't have enough data to process the action, so we want to "skip"
while (next == null)
next = commands.next(rs);
return next;
}
private Action selectRow(LinkedHashMap<Symbol, Object> partition)
{
var builder = select(partition);
List<Clustering<ByteBuffer>> partitions = model.partitions(builder.build());
switch (partitions.size())
{
case 0:
return null;
case 1:
break;
default:
throw new IllegalStateException("Model matched multiple partitions, only 1 is expected");
}
BytesPartitionState state = model.get(partitions.get(0));
if (state == null)
return null;
NavigableSet<Clustering<ByteBuffer>> clusteringKeys = state.clusteringKeys();
if (clusteringKeys.isEmpty())
return null;
Clustering<ByteBuffer> clusteringKey = rs.pickOrderedSet(clusteringKeys);
for (Symbol ck : model.factory.clusteringColumns)
builder.value(ck, clusteringKey.bufferAt(model.factory.clusteringColumns.indexOf(ck)));
return query(builder.build());
}
private Select.Builder select(LinkedHashMap<Symbol, Object> partition)
{
Select.Builder builder = Select.builder().table(metadata);
for (var e : partition.entrySet())
builder.value(e.getKey(), e.getValue());
return builder;
}
private Action query(Select select)
{
return query(select, ConsistencyLevel.ALL, o -> model.validate(RowUtil.toByteBuffer(o), select));
}
private Action query(Mutation mutation)
{
return query(mutation, ConsistencyLevel.QUORUM, o -> model.update(mutation));
}
private Action query(Txn txn)
{
return query(txn, ConsistencyLevel.ALL, o -> model.updateAndValidate(RowUtil.toByteBuffer(o), txn));
}
private String humanReadable(Statement stmt, @Nullable String annotate)
{
// With UTF-8 some chars can cause printing issues leading to error messages that don't reproduce the original issue.
// To avoid this problem, always escape the CQL so nothing gets lost
String cql = StringUtils.escapeControlChars(stmt.visit(StandardVisitors.DEBUG).toCQL(CQLFormatter.None.instance));
if (annotate != null)
cql += " -- " + annotate;
return cql;
}
private Action query(Statement statement, ConsistencyLevel cl, Consumer<Object[][]> onSuccess)
{
// Always use node 1 since we're in single node mode
int nodeId = 1;
String postfix = "on node" + nodeId;
switch (statement.kind())
{
case SELECT: break;
case MUTATION: {
Mutation mutation = statement.asMutation();
if (mutation.isCas())
postfix += ", would apply " + model.shouldApply(mutation);
} break;
case TXN: {
Txn txn = statement.asTxn();
postfix += ", would apply " + model.shouldApply(txn);
} break;
default:
throw new UnsupportedOperationException(statement.kind().name());
}
logger.warn("[step={}] Executing query: {}", step, humanReadable(statement, postfix));
return new SimulatedActionCallable<>(statement.getClass().getSimpleName(),
Action.Modifiers.RELIABLE_NO_TIMEOUTS,
Action.Modifiers.RELIABLE_NO_TIMEOUTS,
simulation.simulated,
simulation.cluster.get(nodeId),
query(statement, cl))
{
final int step_id = step;
final long createdAtNanos = simulation.simulated.time.nanoTime();
@Override
public void accept(Object[][] objects, Throwable throwable)
{
if (throwable != null)
{
logger.error("[step={}] failed", step_id, throwable);
simulated.failures.accept(throwable);
return;
}
logger.warn("[step={}] completed after {}", step_id, Duration.ofNanos(simulation.simulated.time.nanoTime() - createdAtNanos));
onSuccess.accept(objects);
}
};
}
private IIsolatedExecutor.SerializableCallable<Object[][]> query(Statement statement, ConsistencyLevel cl)
{
// Simulator acts differently than jvm-dtest, so ByteBuffer isn't safe!
// java.lang.RuntimeException: java.io.NotSerializableException: java.nio.HeapByteBuffer
// So switch to literals for now
statement = statement.visit(StandardVisitors.BIND_TO_LITERAL);
String cql = statement.toCQL();
Object[] binds = statement.binds();
return () -> {
Query q = new Query(cql, Long.MIN_VALUE, false, cl, null, binds);
return q.call().toObjectArrays();
};
}
}
}

View File

@ -19,7 +19,6 @@
package org.apache.cassandra.simulator.test;
import java.io.IOException;
import java.util.EnumMap;
import java.util.IdentityHashMap;
import org.junit.Test;
@ -28,14 +27,10 @@ import org.apache.cassandra.concurrent.ExecutorFactory;
import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.simulator.ActionList;
import org.apache.cassandra.simulator.Debug;
import org.apache.cassandra.simulator.cluster.ClusterActionListener.NoOpListener;
import org.apache.cassandra.simulator.cluster.ClusterActions;
import org.apache.cassandra.simulator.cluster.ClusterActions.Options;
import org.apache.cassandra.utils.concurrent.CountDownLatch;
import static org.apache.cassandra.simulator.cluster.ClusterActions.InitialConfiguration.initializeAll;
import static org.apache.cassandra.simulator.cluster.ClusterActions.Options.noActions;
public class TrivialSimulationTest extends SimulationTestBase
{
@ -50,14 +45,12 @@ public class TrivialSimulationTest extends SimulationTestBase
public void trivialTest() throws IOException // for demonstration/experiment purposes
{
simulate((simulation) -> {
Options options = noActions(simulation.cluster.size());
ClusterActions clusterActions = new ClusterActions(simulation.simulated, simulation.cluster,
options, new NoOpListener(), new Debug(new EnumMap<>(Debug.Info.class), new int[0]));
ClusterActions clusterActions = ClusterActions.simple(simulation.simulated, simulation.cluster);
return ActionList.of(clusterActions.initializeCluster(initializeAll(simulation.cluster.size())),
simulation.schemaChange(1, "CREATE KEYSPACE ks WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor' : 3}"),
simulation.schemaChange(1, "CREATE TABLE IF NOT EXISTS ks.tbl (pk int PRIMARY KEY, v int)"));
},
(simulation) -> ActionList.of(simulation.executeQuery(1, "INSERT INTO ks.tbl VALUES (1,1)", ConsistencyLevel.QUORUM),
(simulation) -> ActionList.of(simulation.executeQuery(1, "INSERT INTO ks.tbl (pk, v) VALUES (1,1)", ConsistencyLevel.QUORUM),
simulation.executeQuery(1, "SELECT * FROM ks.tbl WHERE pk = 1", ConsistencyLevel.QUORUM)),
(simulation) -> ActionList.of(),
(config) -> config

View File

@ -80,6 +80,7 @@ public class DatabaseDescriptorRefTest
"org.apache.cassandra.config.AccordSpec",
"org.apache.cassandra.config.AccordSpec$JournalSpec",
"org.apache.cassandra.config.AccordSpec$MinEpochRetrySpec",
"org.apache.cassandra.config.AccordSpec$MixedTimeSourceHandling",
"org.apache.cassandra.config.AccordSpec$FetchRetrySpec",
"org.apache.cassandra.config.AccordSpec$TransactionalRangeMigration",
"org.apache.cassandra.config.AccordSpec$QueueShardModel",

View File

@ -60,9 +60,9 @@ public class CQLFormatterPrettyPrintTest
}
@Test
public void updatTest()
public void updateTest()
{
Mutation.UpdateBuilder builder = Mutation.update(TBL1);
Mutation.TableBasedUpdateBuilder builder = Mutation.update(TBL1);
builder.value("pk", 0);
builder.set("v0", 0);
builder.set("v1", 0);
@ -80,7 +80,7 @@ public class CQLFormatterPrettyPrintTest
@Test
public void updateWithInClause()
{
Mutation.UpdateBuilder builder = Mutation.update(TBL1);
Mutation.TableBasedUpdateBuilder builder = Mutation.update(TBL1);
builder.set("v0", 0);
builder.in("pk", Bind.of(42), Literal.of(78));
Assertions.assertThat(builder.build().toCQL(format())).isEqualTo("UPDATE ks.tbl\n" +
@ -93,7 +93,7 @@ public class CQLFormatterPrettyPrintTest
@Test
public void updateWithBetweenClause()
{
Mutation.UpdateBuilder builder = Mutation.update(TBL1);
Mutation.TableBasedUpdateBuilder builder = Mutation.update(TBL1);
builder.set("v0", 0);
builder.value("pk", 0);
builder.between("ck", Bind.of(42), Literal.of(78));
@ -108,7 +108,7 @@ public class CQLFormatterPrettyPrintTest
@Test
public void updateWithIsClause()
{
Mutation.UpdateBuilder builder = Mutation.update(TBL1);
Mutation.TableBasedUpdateBuilder builder = Mutation.update(TBL1);
builder.set("v0", 0);
builder.value("pk", 0);
builder.is(new Symbol("ck", Int32Type.instance),

View File

@ -60,6 +60,11 @@ public interface Conditional extends Expression
return Collections.singletonList(this);
}
static Builder builder()
{
return new Builder();
}
class Where implements Conditional
{
public enum Inequality
@ -86,7 +91,7 @@ public interface Conditional extends Expression
case EQUAL: return rc == 0;
case NOT_EQUAL: return rc != 0;
case GREATER_THAN: return rc > 0;
case GREATER_THAN_EQ: return rc >=0;
case GREATER_THAN_EQ: return rc >= 0;
case LESS_THAN: return rc < 0;
case LESS_THAN_EQ: return rc <=0;
default: throw new UnsupportedOperationException(this.name());
@ -405,7 +410,7 @@ public interface Conditional extends Expression
return in(new Symbol(name, type), values.stream().map(v -> new Bind(v, type)).collect(Collectors.toList()));
}
T is(Symbol ref, Is.Kind kind);
T is(ReferenceExpression ref, Is.Kind kind);
@Override
default T value(Symbol symbol, Expression e)
@ -478,11 +483,16 @@ public interface Conditional extends Expression
}
@Override
public Builder is(Symbol ref, Is.Kind kind)
public Builder is(ReferenceExpression ref, Is.Kind kind)
{
return add(new Is(ref, kind));
}
public Builder is(String ref, Is.Kind kind)
{
return is(Symbol.unknownType(ref), kind);
}
public Conditional build()
{
if (sub.isEmpty())

View File

@ -236,6 +236,23 @@ public class CreateIndexDDL implements Element
}
}
public static class IndexedColumn
{
public final Symbol symbol;
public final CreateIndexDDL indexDDL;
public IndexedColumn(Symbol symbol, CreateIndexDDL indexDDL)
{
this.symbol = symbol;
this.indexDDL = indexDDL;
}
public EnumSet<CreateIndexDDL.QueryType> supportedQueries()
{
return indexDDL.indexer.supportedQueries(symbol.type());
}
}
public static class CollectionReference implements ReferenceExpression
{
public enum Kind { FULL, KEYS, ENTRIES }

View File

@ -91,9 +91,14 @@ public class FunctionCall implements Expression
return new FunctionCall(name, as, returnType);
}
public static FunctionCall writetime(Symbol symbol)
{
return new FunctionCall("writetime", Collections.singletonList(symbol), LongType.instance);
}
public static FunctionCall writetime(String column, AbstractType<?> type)
{
return new FunctionCall("writetime", Collections.singletonList(new Symbol(column, type)), LongType.instance);
return writetime(new Symbol(column, type));
}
public static FunctionCall countStar()

View File

@ -31,8 +31,10 @@ import java.util.function.Function;
import java.util.stream.Stream;
import javax.annotation.Nullable;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.harry.model.DetailedTableMetadata;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
@ -103,9 +105,19 @@ public abstract class Mutation implements Statement
return new InsertBuilder(metadata);
}
public static UpdateBuilder update(TableMetadata metadata)
public static TableBasedUpdateBuilder update(TableMetadata metadata)
{
return new UpdateBuilder(metadata);
return new TableBasedUpdateBuilder(metadata);
}
public static BasicUpdateBuilder update(String table)
{
return new BasicUpdateBuilder(new TableReference(table));
}
public static BasicUpdateBuilder update(String ks, String table)
{
return new BasicUpdateBuilder(new TableReference(Optional.of(ks), table));
}
public static DeleteBuilder delete(TableMetadata metadata)
@ -668,15 +680,15 @@ WHERE PK_column_conditions
}
}
public static abstract class BaseBuilder<T, B extends BaseBuilder<T, B>> implements Conditional.EqBuilderPlus<B>
public static abstract class TableBasedBuilder<T, B extends TableBasedBuilder<T, B>> implements Conditional.EqBuilderPlus<B>
{
private final Kind kind;
protected final TableMetadata metadata;
protected final LinkedHashSet<Symbol> partitionColumns, clusteringColumns, primaryColumns, regularAndStatic, allColumns;
private boolean includeKeyspace = true;
private final Set<Symbol> neededPks = new HashSet<>();
private final PkAware pkAware;
protected BaseBuilder(Kind kind, TableMetadata table)
protected TableBasedBuilder(Kind kind, TableMetadata table)
{
this.kind = kind;
this.metadata = table;
@ -690,7 +702,7 @@ WHERE PK_column_conditions
this.regularAndStatic = new LinkedHashSet<>();
this.regularAndStatic.addAll(toSet(table.regularAndStaticColumns()));
this.allColumns = toSet(table.columnsInFixedOrder());
neededPks.addAll(partitionColumns);
this.pkAware = new PkAware(new DetailedTableMetadata(metadata));
}
protected Symbol find(String name)
@ -708,36 +720,12 @@ WHERE PK_column_conditions
protected void assertAllPksHaveEq()
{
if (neededPks.isEmpty())
return;
throw new IllegalStateException("Attempted to create a " + kind + " but not all partition columns have an equality condition; missing " + neededPks);
pkAware.assertAllPksHaveEq();
}
protected void maybePkEq(Expression symbol)
{
if (symbol instanceof Symbol)
pkEq((Symbol) symbol);
}
private void pkEq(Symbol symbol)
{
neededPks.remove(symbol);
}
public B includeKeyspace(boolean value)
{
this.includeKeyspace = value;
return (B) this;
}
public B includeKeyspace()
{
return includeKeyspace(true);
}
public B excludeKeyspace()
{
return includeKeyspace(false);
pkAware.maybePkEq(symbol);
}
protected TableReference tableRef()
@ -746,7 +734,7 @@ WHERE PK_column_conditions
}
}
public static class InsertBuilder extends BaseBuilder<Insert, InsertBuilder>
public static class InsertBuilder extends TableBasedBuilder<Insert, InsertBuilder>
{
private final LinkedHashMap<Symbol, Expression> values = new LinkedHashMap<>();
private boolean ifNotExists = false;
@ -805,131 +793,121 @@ WHERE PK_column_conditions
}
}
public static class UpdateBuilder extends BaseBuilder<Update, UpdateBuilder> implements Conditional.ConditionalBuilderPlus<UpdateBuilder>
public interface WithTTl<B extends WithTTl<B>>
{
B ttl(Value value);
default B ttl(int value)
{
return ttl(Bind.of(value));
}
}
public interface WithTimestamp<B extends WithTimestamp<B>>
{
default B timestamp(long value)
{
return timestamp(Literal.of(value));
}
B timestamp(Value value);
}
public interface UpdateBuilder<B extends UpdateBuilder<B>> extends Conditional.ConditionalBuilder<B>,
Conditional.EqBuilder<B>,
WithTTl<B>, WithTimestamp<B>
{
B set(Symbol column, Expression value);
default B set(String column, Object value, AbstractType<?> type)
{
return set(new Symbol(column, type), new Literal(value, type));
}
default B ifExists()
{
return ifCondition(CasCondition.Simple.Exists);
}
B ifCondition(CasCondition condition);
Update build();
}
@SuppressWarnings("unchecked")
public static class BaseUpdateBuilder<B extends BaseUpdateBuilder<B>> implements UpdateBuilder<B>
{
private TableReference table;
private @Nullable TTL ttl;
private @Nullable Timestamp timestamp;
private final LinkedHashMap<Symbol, Expression> set = new LinkedHashMap<>();
private final Conditional.Builder where = new Conditional.Builder();
private @Nullable CasCondition casCondition;
protected UpdateBuilder(TableMetadata table)
public BaseUpdateBuilder(TableReference table)
{
super(Kind.UPDATE, table);
this.table = table;
}
public UpdateBuilder timestamp(long value)
{
return timestamp(Literal.of(value));
}
public UpdateBuilder timestamp(Value value)
@Override
public B timestamp(Value value)
{
this.timestamp = new Timestamp(value);
return this;
return (B) this;
}
public UpdateBuilder ttl(Value value)
@Override
public B ttl(Value value)
{
this.ttl = new TTL(value);
return this;
return (B) this;
}
public UpdateBuilder ttl(int value)
{
return ttl(Bind.of(value));
}
public UpdateBuilder ifExists()
{
casCondition = CasCondition.Simple.Exists;
return this;
}
public UpdateBuilder ifCondition(CasCondition condition)
@Override
public B ifCondition(CasCondition condition)
{
casCondition = condition;
return this;
return (B) this;
}
public UpdateBuilder set(Symbol column, Expression value)
@Override
public B set(Symbol column, Expression value)
{
if (!regularAndStatic.contains(column))
throw new IllegalArgumentException("Attempted to set a non regular or static column " + column + "; expected " + regularAndStatic);
set.put(column, value);
return this;
}
public UpdateBuilder set(String column, int value)
{
Symbol symbol = find(column);
if (!symbol.type().equals(Int32Type.instance))
throw new AssertionError("Expected int type but given " + symbol.type().asCQL3Type());
return set(symbol, Bind.of(value));
}
public UpdateBuilder set(String column, Object value)
{
Symbol symbol = find(column);
return set(symbol, new Bind(value, symbol.type()));
}
public UpdateBuilder set(String column, Expression expression)
{
return set(find(column), expression);
}
public UpdateBuilder set(String column, Function<Symbol, Expression> fn)
{
Symbol symbol = find(column);
return set(symbol, fn.apply(symbol));
}
public UpdateBuilder set(String column, String value)
{
Symbol symbol = find(column);
return set(symbol, new Bind(symbol.type().asCQL3Type().fromCQLLiteral(value), symbol.type()));
return (B) this;
}
@Override
public UpdateBuilder where(Expression ref, Conditional.Where.Inequality kind, Expression expression)
public B where(Expression ref, Conditional.Where.Inequality kind, Expression expression)
{
if (kind == Conditional.Where.Inequality.EQUAL)
maybePkEq(ref);
where.where(ref, kind, expression);
return this;
return (B) this;
}
@Override
public UpdateBuilder between(Expression ref, Expression start, Expression end)
public B between(Expression ref, Expression start, Expression end)
{
where.between(ref, start, end);
return this;
return (B) this;
}
@Override
public UpdateBuilder in(ReferenceExpression ref, List<? extends Expression> expressions)
public B in(ReferenceExpression ref, List<? extends Expression> expressions)
{
maybePkEq(ref);
where.in(ref, expressions);
return this;
return (B) this;
}
@Override
public UpdateBuilder is(Symbol ref, Conditional.Is.Kind kind)
public B is(ReferenceExpression ref, Conditional.Is.Kind kind)
{
where.is(ref, kind);
return this;
return (B) this;
}
@Override
public Update build()
{
assertAllPksHaveEq();
if (set.isEmpty())
throw new IllegalStateException("Unable to create an Update without a SET section; set function was never called");
return new Update(tableRef(),
return new Update(table,
(ttl == null && timestamp == null) ? Optional.empty() : Optional.of(new Using(Optional.ofNullable(ttl), Optional.ofNullable(timestamp))),
new LinkedHashMap<>(set),
where.build(),
@ -937,7 +915,126 @@ WHERE PK_column_conditions
}
}
public static class DeleteBuilder extends BaseBuilder<Delete, DeleteBuilder> implements Conditional.ConditionalBuilderPlus<DeleteBuilder>
public static class BasicUpdateBuilder extends BaseUpdateBuilder<BasicUpdateBuilder>
{
public BasicUpdateBuilder(TableReference table)
{
super(table);
}
}
private static class PkAware
{
private final Set<Symbol> neededPks = new HashSet<>();
private final DetailedTableMetadata metadata;
private PkAware(DetailedTableMetadata metadata)
{
this.metadata = metadata;
neededPks.addAll(metadata.partitionColumns);
}
protected void assertAllPksHaveEq()
{
if (neededPks.isEmpty())
return;
throw new IllegalStateException("Attempted to create a mutation but not all partition columns have an equality condition; missing " + neededPks);
}
protected void maybePkEq(Expression symbol)
{
if (symbol instanceof Symbol)
pkEq((Symbol) symbol);
}
private void pkEq(Symbol symbol)
{
neededPks.remove(symbol);
}
}
public static class TableBasedUpdateBuilder extends BaseUpdateBuilder<TableBasedUpdateBuilder>
implements UpdateBuilder<TableBasedUpdateBuilder>, Conditional.ConditionalBuilderPlus<TableBasedUpdateBuilder>
{
private final DetailedTableMetadata metadata;
private final PkAware pkAware;
protected TableBasedUpdateBuilder(TableMetadata table)
{
super(TableReference.from(table));
this.metadata = new DetailedTableMetadata(table);
this.pkAware = new PkAware(metadata);
}
@Override
public TableBasedUpdateBuilder set(Symbol column, Expression value)
{
if (!metadata.regularAndStaticColumns.contains(column))
throw new IllegalArgumentException("Attempted to set a non regular or static column " + column + "; expected " + metadata.regularAndStaticColumns);
return super.set(column, value);
}
public TableBasedUpdateBuilder set(String column, int value)
{
Symbol symbol = metadata.find(column);
if (!symbol.type().equals(Int32Type.instance))
throw new AssertionError("Expected int type but given " + symbol.type().asCQL3Type());
return set(symbol, Bind.of(value));
}
public TableBasedUpdateBuilder set(String column, Object value)
{
Symbol symbol = metadata.find(column);
return set(symbol, new Bind(value, symbol.type()));
}
public TableBasedUpdateBuilder set(String column, Expression expression)
{
return set(metadata.find(column), expression);
}
public TableBasedUpdateBuilder set(String column, Function<Symbol, Expression> fn)
{
Symbol symbol = metadata.find(column);
return set(symbol, fn.apply(symbol));
}
public TableBasedUpdateBuilder set(String column, String value)
{
Symbol symbol = metadata.find(column);
return set(symbol, new Bind(symbol.type().asCQL3Type().fromCQLLiteral(value), symbol.type()));
}
@Override
public TableBasedUpdateBuilder where(Expression ref, Conditional.Where.Inequality kind, Expression expression)
{
if (kind == Conditional.Where.Inequality.EQUAL)
pkAware.maybePkEq(ref);
return super.where(ref, kind, expression);
}
@Override
public TableBasedUpdateBuilder in(ReferenceExpression ref, List<? extends Expression> expressions)
{
pkAware.maybePkEq(ref);
return super.in(ref, expressions);
}
@Override
public Update build()
{
pkAware.assertAllPksHaveEq();
return super.build();
}
@Override
public TableMetadata metadata()
{
return metadata.metadata;
}
}
public static class DeleteBuilder extends TableBasedBuilder<Delete, DeleteBuilder> implements Conditional.ConditionalBuilderPlus<DeleteBuilder>
{
private final List<Symbol> columns = new ArrayList<>();
private @Nullable Timestamp timestamp = null;
@ -1033,7 +1130,7 @@ WHERE PK_column_conditions
}
@Override
public DeleteBuilder is(Symbol ref, Conditional.Is.Kind kind)
public DeleteBuilder is(ReferenceExpression ref, Conditional.Is.Kind kind)
{
where.is(ref, kind);
return this;

View File

@ -408,7 +408,7 @@ FROM [keyspace_name.] table_name
}
@Override
public T is(Symbol ref, Conditional.Is.Kind kind)
public T is(ReferenceExpression ref, Conditional.Is.Kind kind)
{
where.is(ref, kind);
return (T) this;

View File

@ -24,6 +24,7 @@ import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import accord.utils.Invariants;
import org.apache.cassandra.utils.ByteBufferUtil;
public interface Statement extends Element
@ -31,6 +32,27 @@ public interface Statement extends Element
enum Kind { SELECT, MUTATION, TXN }
Kind kind();
default Mutation asMutation()
{
if (kind() != Kind.MUTATION)
throw Invariants.illegalState("Unable to return Mutation; kind=%s", kind());
return (Mutation) this;
}
default Select asSelect()
{
if (kind() != Kind.SELECT)
throw Invariants.illegalState("Unable to return Select; kind=%s", kind());
return (Select) this;
}
default Txn asTxn()
{
if (kind() != Kind.TXN)
throw Invariants.illegalState("Unable to return Txn; kind=%s", kind());
return (Txn) this;
}
default Object[] binds()
{
return streamRecursive()

View File

@ -156,7 +156,7 @@ public class Txn implements Statement
Statement update = m.visit(v);
if (!(update instanceof Mutation))
throw new IllegalArgumentException("Unable to use type " + update.getClass() + " where Mutation is expected");
updated |= update != m;
localUpdated |= update != m;
mutations.add((Mutation) update);
}
ifBlock = !localUpdated ? this.ifBlock : Optional.of(new If(c, mutations));
@ -335,8 +335,8 @@ public class Txn implements Statement
public static class If implements Element
{
private final Conditional conditional;
private final List<Mutation> mutations;
public final Conditional conditional;
public final List<Mutation> mutations;
public If(Conditional conditional, List<Mutation> mutations)
{

View File

@ -48,6 +48,7 @@ import org.apache.cassandra.net.MockMessagingSpy;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.PreserveTimestamp;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.accord.AccordResult;
import org.apache.cassandra.service.accord.AccordTestUtils;
@ -234,7 +235,7 @@ public class HintsServiceTest
}
@Override
public IAccordResult<TxnResult> mutateWithAccordAsync(ClusterMetadata cm, Mutation mutation, @Nullable ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime)
public IAccordResult<TxnResult> mutateWithAccordAsync(ClusterMetadata cm, Mutation mutation, @Nullable ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime, PreserveTimestamp preserveTimestamps)
{
accordTxnCount.incrementAndGet();
TxnId txnId = AccordTestUtils.txnId(42, 43, 44);

View File

@ -0,0 +1,81 @@
/*
* 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.service;
import org.junit.Test;
import static org.apache.cassandra.service.TimestampSource.merge;
import static org.junit.Assert.assertEquals;
public class TimestampSourceTest
{
@Test
public void testMergeSameValues()
{
for (var v : TimestampSource.values())
assertEquals(v, merge(v, v));
}
@Test
public void testMergeAssociativity()
{
for (var left : TimestampSource.values())
{
for (var right : TimestampSource.values())
{
assertEquals(merge(left, right), merge(right, left));
}
}
}
@Test
public void testMergeIdentityWithUnknown()
{
// unknown should act as identity element
assertEquals(TimestampSource.server, merge(TimestampSource.server, TimestampSource.unknown));
assertEquals(TimestampSource.using, merge(TimestampSource.using, TimestampSource.unknown));
assertEquals(TimestampSource.mixed, merge(TimestampSource.mixed, TimestampSource.unknown));
}
@Test
public void testMergeAbsorbingElementWithMixed()
{
for (var v : TimestampSource.values())
assertEquals(TimestampSource.mixed, merge(TimestampSource.mixed, v));
for (var v : TimestampSource.values())
assertEquals(TimestampSource.mixed, merge(v, TimestampSource.mixed));
}
@Test
public void testMergeServerWithOthers()
{
assertEquals(TimestampSource.server, merge(TimestampSource.server, TimestampSource.unknown));
assertEquals(TimestampSource.mixed, merge(TimestampSource.server, TimestampSource.using));
assertEquals(TimestampSource.mixed, merge(TimestampSource.server, TimestampSource.mixed));
}
@Test
public void testMergeUsingWithOthers()
{
assertEquals(TimestampSource.using, merge(TimestampSource.using, TimestampSource.unknown));
assertEquals(TimestampSource.mixed, merge(TimestampSource.using, TimestampSource.server));
assertEquals(TimestampSource.mixed, merge(TimestampSource.using, TimestampSource.mixed));
}
}

View File

@ -18,6 +18,7 @@
package org.apache.cassandra.utils;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@ -36,17 +37,21 @@ import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import javax.annotation.Nullable;
import com.google.common.collect.Iterables;
import accord.utils.Gens;
import accord.utils.RandomSource;
import org.apache.cassandra.cql3.KnownIssue;
import org.apache.cassandra.cql3.ast.AssignmentOperator;
import org.apache.cassandra.cql3.ast.Bind;
import org.apache.cassandra.cql3.ast.CasCondition;
import org.apache.cassandra.cql3.ast.Conditional;
import org.apache.cassandra.cql3.ast.CreateIndexDDL;
import org.apache.cassandra.cql3.ast.Expression;
import org.apache.cassandra.cql3.ast.Literal;
import org.apache.cassandra.cql3.ast.Mutation;
@ -66,6 +71,8 @@ import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.db.marshal.MapType;
import org.apache.cassandra.db.marshal.SetType;
import org.apache.cassandra.db.marshal.ShortType;
import org.apache.cassandra.harry.model.ASTSingleTableModel;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.quicktheories.core.Gen;
@ -73,11 +80,33 @@ import org.quicktheories.core.RandomnessSource;
import org.quicktheories.generators.SourceDSL;
import org.quicktheories.impl.Constraint;
import static org.apache.cassandra.utils.AbstractTypeGenerators.getTypeSupport;
import static org.apache.cassandra.utils.Generators.SYMBOL_GEN;
public class ASTGenerators
{
public static final EnumSet<KnownIssue> IGNORE_ISSUES = KnownIssue.ignoreAll();
public static final EnumSet<KnownIssue> IGNORED_ISSUES = KnownIssue.ignoreAll();
public static Gen<LinkedHashMap<Symbol, Object>> columnValues(List<Symbol> columns)
{
List<Gen<?>> gens = new ArrayList<>(columns.size());
for (int i = 0; i < columns.size(); i++)
gens.add(getTypeSupport(columns.get(i).type()).valueGen);
return rs -> {
LinkedHashMap<Symbol, Object> vs = new LinkedHashMap<>();
for (int i = 0; i < columns.size(); i++)
vs.put(columns.get(i), gens.get(i).generate(rs));
return vs;
};
}
public static Select select(TableMetadata metadata, LinkedHashMap<Symbol, Object> map)
{
Select.TableBasedBuilder builder = Select.builder(metadata);
for (var e : map.entrySet())
builder.value(e.getKey(), e.getValue());
return builder.build();
}
static Gen<Value> valueGen(Object value, AbstractType<?> type)
{
@ -87,7 +116,7 @@ public class ASTGenerators
static Gen<Value> valueGen(AbstractType<?> type)
{
Gen<?> v = AbstractTypeGenerators.getTypeSupport(type).valueGen;
Gen<?> v = getTypeSupport(type).valueGen;
return rnd -> valueGen(v.generate(rnd), type).generate(rnd);
}
@ -151,7 +180,7 @@ public class ASTGenerators
//NOTE: see https://the-asf.slack.com/archives/CK23JSY2K/p1724819303058669 - varint didn't fail but serialized using int32 which causes equality mismatches for pk/ck lookups
if ((e.type().unwrap() == ShortType.instance
|| e.type().unwrap() == IntegerType.instance)
&& IGNORE_ISSUES.contains(KnownIssue.SHORT_AND_VARINT_GET_INT_FUNCTIONS)) // seed=7525457176675272023L
&& IGNORED_ISSUES.contains(KnownIssue.SHORT_AND_VARINT_GET_INT_FUNCTIONS)) // seed=7525457176675272023L
{
left = new TypeHint(left);
right = new TypeHint(right);
@ -173,7 +202,7 @@ public class ASTGenerators
public ExpressionBuilder(AbstractType<?> type)
{
this.type = type.unwrap();
this.valueGen = AbstractTypeGenerators.getTypeSupport(this.type).valueGen;
this.valueGen = getTypeSupport(this.type).valueGen;
this.allowedOperators = Operator.supportsOperators(this.type);
}
@ -184,6 +213,12 @@ public class ASTGenerators
return this;
}
public ExpressionBuilder disallowEmpty()
{
useEmpty = i -> false;
return this;
}
public ExpressionBuilder allowNull()
{
useNull = SourceDSL.integers().between(1, 100).map(i -> i < 10);
@ -287,7 +322,7 @@ public class ASTGenerators
return this;
}
public SelectGenBuilder withKeys(Gen<Map<Symbol, Object>> partitionKeys, Gen<Map<Symbol, Object>> clusteringKeys)
public SelectGenBuilder withKeys(Gen<? extends Map<Symbol, Object>> partitionKeys, Gen<? extends Map<Symbol, Object>> clusteringKeys)
{
keyGen = rs -> {
Map<Symbol, Expression> keys = new LinkedHashMap<>();
@ -345,7 +380,7 @@ public class ASTGenerators
{
Map<ColumnMetadata, Gen<?>> gens = new LinkedHashMap<>();
for (ColumnMetadata col : metadata.columnsInFixedOrder())
gens.put(col, AbstractTypeGenerators.getTypeSupport(col.type).valueGen);
gens.put(col, getTypeSupport(col.type).valueGen);
return rnd -> {
Map<Symbol, Expression> output = new LinkedHashMap<>();
for (ColumnMetadata col : metadata.partitionKeyColumns())
@ -377,7 +412,7 @@ public class ASTGenerators
private boolean allowPartitionOnlyUpdate = true;
private boolean allowPartitionOnlyInsert = true;
private boolean allowUpdateMultipleClusteringKeys = true;
private EnumSet<KnownIssue> ignoreIssues = IGNORE_ISSUES;
private EnumSet<KnownIssue> ignoreIssues = IGNORED_ISSUES;
public MutationGenBuilder(TableMetadata metadata)
{
@ -433,6 +468,18 @@ public class ASTGenerators
return this;
}
public MutationGenBuilder disallowEmpty()
{
columnExpressions.values().forEach(ExpressionBuilder::disallowEmpty);
return this;
}
public MutationGenBuilder disallowEmpty(Symbol symbol)
{
columnExpressions.get(symbol).disallowEmpty();
return this;
}
public MutationGenBuilder allowNull(Symbol symbol)
{
columnExpressions.get(symbol).allowNull();
@ -456,6 +503,13 @@ public class ASTGenerators
return this;
}
public MutationGenBuilder withTxnSafe()
{
return withoutTimestamp()
.withoutTtl()
.withoutTransaction();
}
public MutationGenBuilder withoutTransaction()
{
withoutCas();
@ -658,7 +712,7 @@ public class ASTGenerators
}
case UPDATE:
{
Mutation.UpdateBuilder builder = Mutation.update(metadata);
Mutation.TableBasedUpdateBuilder builder = Mutation.update(metadata);
var ttl = ttlGen.generate(rnd);
if (ttl.isPresent())
builder.ttl(valueGen(ttl.getAsInt(), Int32Type.instance).generate(rnd));
@ -673,7 +727,7 @@ public class ASTGenerators
if (!staticColumns.isEmpty() && allowPartitionOnlyUpdate && bool.generate(rnd))
{
var columnsToGenerate = new LinkedHashSet<>(subset(rnd, staticColumns));
Conditional.EqBuilder<Mutation.UpdateBuilder> setBuilder = builder::set;
Conditional.EqBuilder<Mutation.TableBasedUpdateBuilder> setBuilder = builder::set;
generateRemaining(rnd, bool, Mutation.Kind.UPDATE, isTransaction, typeToReference, setBuilder, columnsToGenerate);
if (isCas)
@ -719,7 +773,7 @@ public class ASTGenerators
if (!staticColumns.isEmpty() && bool.generate(rnd))
columnsToGenerate.addAll(subset(rnd, staticColumns));
}
Conditional.EqBuilder<Mutation.UpdateBuilder> setBuilder = builder::set;
Conditional.EqBuilder<Mutation.TableBasedUpdateBuilder> setBuilder = builder::set;
generateRemaining(rnd, bool, Mutation.Kind.UPDATE, isTransaction, typeToReference, setBuilder, columnsToGenerate);
return builder.build();
}
@ -1001,9 +1055,7 @@ public class ASTGenerators
builder.addReturn(selectGen.generate(rnd));
}
MutationGenBuilder mutationBuilder = new MutationGenBuilder(metadata)
.withoutCas()
.withoutTimestamp()
.withoutTtl()
.withTxnSafe()
.withAllowUpdateMultipleClusteringKeys(false)
.withReferences(new ArrayList<>(builder.allowedReferences()));
if (!allowReferences)
@ -1049,11 +1101,145 @@ public class ASTGenerators
private static Gen<Conditional.Where> whereGen(Reference ref)
{
Gen<Conditional.Where.Inequality> kindGen = SourceDSL.arbitrary().enumValues(Conditional.Where.Inequality.class);
Gen<?> dataGen = AbstractTypeGenerators.getTypeSupport(ref.type()).valueGen;
Gen<?> dataGen = getTypeSupport(ref.type()).valueGen;
return rnd -> {
Conditional.Where.Inequality kind = kindGen.generate(rnd);
return Conditional.Where.create(kind, ref, valueGen(dataGen.generate(rnd), ref.type()).generate(rnd));
};
}
}
public static MutationGenBuilder mutationBuilder(RandomSource rs,
ASTSingleTableModel model,
List<LinkedHashMap<Symbol, Object>> uniquePartitions,
Function<Symbol, CreateIndexDDL.IndexedColumn> indexes)
{
return mutationBuilder(IGNORED_ISSUES, rs, model, uniquePartitions, indexes);
}
public static MutationGenBuilder mutationBuilder(EnumSet<KnownIssue> ignoredIssues,
RandomSource rs,
ASTSingleTableModel model,
List<LinkedHashMap<Symbol, Object>> uniquePartitions,
Function<Symbol, CreateIndexDDL.IndexedColumn> indexes)
{
var boolDistribution = Gens.bools().mixedDistribution();
TableMetadata metadata = model.factory.metadata;
MutationGenBuilder builder = new MutationGenBuilder(metadata)
.withTxnSafe()
.withPartitions(uniquePartitions.size() == 1
? SourceDSL.arbitrary().constant(uniquePartitions.get(0))
: Generators.fromGen(Gens.mixedDistribution(uniquePartitions).next(rs)))
.withColumnExpressions(e -> e.withOperators(Generators.fromGen(boolDistribution.next(rs))))
.withIgnoreIssues(ignoredIssues);
if (ignoredIssues.contains(KnownIssue.SAI_EMPTY_TYPE))
{
model.factory.regularAndStaticColumns.stream()
// exclude SAI indexed columns
.filter(s -> indexes.apply(s) == null || indexes.apply(s).indexDDL.indexer != CreateIndexDDL.SAI)
.forEach(builder::allowEmpty);
}
else
{
model.factory.regularAndStaticColumns.forEach(builder::allowEmpty);
}
model.factory.regularAndStaticColumns.forEach(builder::allowNull);
return builder;
}
public static class ModelBasedTxnGenBuilder
{
private final RandomSource rs; //TODO (now): refactor so ASTGenerators no longer is quicktheories, its too annoying to go back and forth...
private final TableMetadata metadata;
private final ASTSingleTableModel model;
private final Function<Symbol, CreateIndexDDL.IndexedColumn> indexes;
private final Gen<LinkedHashMap<Symbol, Object>> partitionKeyValuesGen;
private Gen<Boolean> bindOrLiteralGen = SourceDSL.booleans().all();
private boolean allowEmpty = true;
public ModelBasedTxnGenBuilder(RandomSource rs,
ASTSingleTableModel model,
Function<Symbol, CreateIndexDDL.IndexedColumn> indexes,
Gen<LinkedHashMap<Symbol, Object>> partitionKeyValuesGen)
{
this.rs = rs;
this.metadata = model.factory.metadata;
this.model = model;
this.indexes = indexes;
this.partitionKeyValuesGen = partitionKeyValuesGen;
}
public ModelBasedTxnGenBuilder disallowEmpty()
{
allowEmpty = false;
return this;
}
public ModelBasedTxnGenBuilder withBindOrLiteralGen(Gen<Boolean> bindOrLiteralGen)
{
this.bindOrLiteralGen = bindOrLiteralGen;
return this;
}
private Gen<Mutation> mutationGen(RandomSource rs, LinkedHashMap<Symbol, Object> pk)
{
MutationGenBuilder builder = mutationBuilder(IGNORED_ISSUES, rs, model, List.of(pk), indexes);
builder.withTxnSafe()
//TODO (now, coverage): remove this as we should support
// working around the bug to make progress
.withAllowUpdateMultipleClusteringKeys(false);
if (!allowEmpty)
builder.disallowEmpty();
return builder.build();
}
private Value value(RandomnessSource rs, ByteBuffer bb, AbstractType<?> type)
{
return bindOrLiteralGen.generate(rs) ? new Bind(bb, type) : new Literal(bb, type);
}
public Gen<Txn> build()
{
Gen<Boolean> boolGen = SourceDSL.booleans().all();
return rnd -> {
var pk = partitionKeyValuesGen.generate(rnd);
var mutation = mutationGen(rs, pk).generate(rnd);
Select select = select(metadata, pk).withLimit(1);
var columns = model.columns(select);
Txn.Builder builder = Txn.builder();
builder.addLet("r1", select);
Reference ref = Reference.of(Symbol.unknownType("r1"));
builder.addReturn(select(metadata, pk));
Conditional.Builder condition = Conditional.builder();
for (var col : columns)
{
if (boolGen.generate(rnd)) continue;
Reference colRef = ref.add(col);
if (boolGen.generate(rnd))
condition.is(colRef, SourceDSL.arbitrary().enumValues(Conditional.Is.Kind.class).generate(rnd));
if (boolGen.generate(rnd))
{
Expression lhs = colRef;
Expression rhs = value(rnd, getTypeSupport(lhs.type()).bytesGen().generate(rnd), lhs.type());
if (boolGen.generate(rnd))
{
var tmp = lhs;
lhs = rhs;
rhs = tmp;
}
Conditional.Where.Inequality inequality = SourceDSL.arbitrary().enumValues(Conditional.Where.Inequality.class).generate(rnd);
condition.where(lhs, inequality, rhs);
}
}
if (condition.isEmpty())
condition.is("r1", Conditional.Is.Kind.NotNull);
builder.addIf(condition.build(), mutation);
return builder.build();
};
}
}
}