diff --git a/CHANGES.txt b/CHANGES.txt index 2b3ab7a8df..69b2ecae81 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/doc/modules/cassandra/pages/architecture/cql-on-accord.adoc b/doc/modules/cassandra/pages/architecture/cql-on-accord.adoc index 91a8fc2854..4ba46c1e79 100644 --- a/doc/modules/cassandra/pages/architecture/cql-on-accord.adoc +++ b/doc/modules/cassandra/pages/architecture/cql-on-accord.adoc @@ -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 diff --git a/doc/modules/cassandra/pages/managing/operating/onboarding-to-accord.adoc b/doc/modules/cassandra/pages/managing/operating/onboarding-to-accord.adoc index 17d4515000..e9fe5e386f 100644 --- a/doc/modules/cassandra/pages/managing/operating/onboarding-to-accord.adoc +++ b/doc/modules/cassandra/pages/managing/operating/onboarding-to-accord.adoc @@ -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. diff --git a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java index 4c19d1c010..5be64ae884 100644 --- a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java +++ b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java @@ -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) diff --git a/src/java/org/apache/cassandra/config/AccordSpec.java b/src/java/org/apache/cassandra/config/AccordSpec.java index ec70398f64..79e6d62211 100644 --- a/src/java/org/apache/cassandra/config/AccordSpec.java +++ b/src/java/org/apache/cassandra/config/AccordSpec.java @@ -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; diff --git a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java index 68b2885e29..c2229b051c 100644 --- a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java @@ -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 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 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); @@ -463,14 +472,30 @@ public class BatchStatement implements CQLStatement.CompositeCQLStatement if (updatesVirtualTables) executeInternalWithoutCondition(queryState, options, requestTime); - else - executeWithoutConditions(getMutations(clientState, options, false, timestamp, nowInSeconds, requestTime), - options.getConsistency(), requestTime); + else + { + PreserveTimestamp preserveTimestamp; + TimestampSource.Collector collector; + if (attrs.isTimestampSet()) + { + preserveTimestamp = PreserveTimestamp.yes; + collector = null; + } + else + { + preserveTimestamp = PreserveTimestamp.no; + collector = new TimestampSource.Collector(); + } + List 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 mutations, ConsistencyLevel cl, Dispatcher.RequestTime requestTime) throws RequestExecutionException, RequestValidationException + private void executeWithoutConditions(List 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; } diff --git a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java index eeadd10456..cc1680b19a 100644 --- a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java +++ b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java @@ -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() diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index 90671c1bd6..90cdf32f36 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -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); diff --git a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java index 67a796a28a..6ca18b9f02 100644 --- a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java @@ -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 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)); } diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 4cc86d4889..5b39dd42da 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -2229,6 +2229,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner CompactionManager.instance.performMaximal(this, splitOutput, permittedParallelism); } + @VisibleForTesting + public List> submitMajorCompaction(boolean splitOutput, int permittedParallelism) + { + return CompactionManager.instance.submitMaximal(this, splitOutput, permittedParallelism); + } + @Override public void forceCompactionForTokenRange(Collection> tokenRanges) throws ExecutionException, InterruptedException { diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index a36d0fc49b..3ab00acfb9 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -1141,6 +1141,11 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan FBUtilities.waitOnFutures(submitMaximal(cfStore, getDefaultGcBefore(cfStore, FBUtilities.nowInSeconds()), splitOutput, permittedParallelism)); } + public List> submitMaximal(final ColumnFamilyStore cfStore, boolean splitOutput, int permittedParallelism) + { + return submitMaximal(cfStore, getDefaultGcBefore(cfStore, FBUtilities.nowInSeconds()), splitOutput, permittedParallelism); + } + public List> submitMaximal(final ColumnFamilyStore cfStore, final long gcBefore, boolean splitOutput, int permittedParallelism) { return submitMaximal(cfStore, gcBefore, splitOutput, permittedParallelism, OperationType.MAJOR_COMPACTION); diff --git a/src/java/org/apache/cassandra/hints/HintsDispatcher.java b/src/java/org/apache/cassandra/hints/HintsDispatcher.java index 350b7445e1..3753b9334d 100644 --- a/src/java/org/apache/cassandra/hints/HintsDispatcher.java +++ b/src/java/org/apache/cassandra/hints/HintsDispatcher.java @@ -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; diff --git a/src/java/org/apache/cassandra/service/PreserveTimestamp.java b/src/java/org/apache/cassandra/service/PreserveTimestamp.java new file mode 100644 index 0000000000..a33b7db9e6 --- /dev/null +++ b/src/java/org/apache/cassandra/service/PreserveTimestamp.java @@ -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); + } +} diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 1c91a1886d..72692af133 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -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 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)mutations, consistencyLevel, updatesView, requestTime); else - dispatchMutationsWithRetryOnDifferentSystem(mutations, consistencyLevel, requestTime); + dispatchMutationsWithRetryOnDifferentSystem(mutations, consistencyLevel, requestTime, preserveTimestamps); } - public static void dispatchMutationsWithRetryOnDifferentSystem(List mutations, ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime) + public static void dispatchMutationsWithRetryOnDifferentSystem(List mutations, ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime, PreserveTimestamp preserveTimestamps) { while (true) { ClusterMetadata cm = ClusterMetadata.current(); try { - SplitMutations splitMutations = splitMutationsIntoAccordAndNormal(cm, (List)mutations); + SplitMutations splitMutations = splitMutationsIntoAccordAndNormal(cm, (List)mutations); List accordMutations = splitMutations.accordMutations(); - IAccordResult accordResult = accordMutations != null ? mutateWithAccordAsync(cm, accordMutations, consistencyLevel, requestTime) : null; List 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 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 accordResult = !accordMutations.isEmpty() ? mutateWithAccordAsync(cm, accordMutations, consistencyLevel, requestTime) : null; + IAccordResult accordResult = !accordMutations.isEmpty() ? mutateWithAccordAsync(cm, accordMutations, consistencyLevel, requestTime, PreserveTimestamp.yes) : null; Throwable failure = null; try diff --git a/src/java/org/apache/cassandra/service/TimestampSource.java b/src/java/org/apache/cassandra/service/TimestampSource.java new file mode 100644 index 0000000000..73d581182e --- /dev/null +++ b/src/java/org/apache/cassandra/service/TimestampSource.java @@ -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()"); + } + } +} diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java index ef4730e818..aef4342c67 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java @@ -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 fragments, TxnCondition condition, @Nullable ConsistencyLevel cassandraCommitCL, boolean preserveTimestamps) + public TxnUpdate(TableMetadatas tables, List 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 diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java b/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java index a113899049..8008df78ab 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java @@ -447,7 +447,7 @@ public class TxnWrite extends AbstractKeySorted 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))); diff --git a/src/java/org/apache/cassandra/service/consensus/migration/ConsensusMigrationMutationHelper.java b/src/java/org/apache/cassandra/service/consensus/migration/ConsensusMigrationMutationHelper.java index b3291994db..c319daee9b 100644 --- a/src/java/org/apache/cassandra/service/consensus/migration/ConsensusMigrationMutationHelper.java +++ b/src/java/org/apache/cassandra/service/consensus/migration/ConsensusMigrationMutationHelper.java @@ -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 mutateWithAccordAsync(ClusterMetadata cm, Mutation mutation, @Nullable ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime) + public IAccordResult 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 mutateWithAccordAsync(ClusterMetadata cm, Collection mutations, @Nullable ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime) + public static IAccordResult mutateWithAccordAsync(ClusterMetadata cm, + Collection 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); } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Query.java b/test/distributed/org/apache/cassandra/distributed/impl/Query.java index 26428034e6..bb67733840 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Query.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Query.java @@ -43,14 +43,21 @@ public class Query implements IIsolatedExecutor.SerializableCallable 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 specs, List> rows) + { + return toObjects(specs, rows, true); + } + + public static Object[][] toObjects(List specs, List> 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 toObjects(ResultSet rs) { return Iterators.transform(rs.iterator(), (Row row) -> { @@ -116,15 +150,15 @@ public class RowUtil public static Iterator toIter(List columnSpecs, Iterator rs) { Iterator> iter = Iterators.transform(rs, - (row) -> { - List 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 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)); } diff --git a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java index 40b76a7a9b..a9fe78f642 100644 --- a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java +++ b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java @@ -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 cluster, int[] cmsNodes) + public static Epoch maxEpoch(ICluster 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 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; } diff --git a/test/distributed/org/apache/cassandra/distributed/test/cql3/AccordInteropMultiNodeTableWalkBase.java b/test/distributed/org/apache/cassandra/distributed/test/cql3/AccordInteropMultiNodeTableWalkBase.java index 6e4c1ad1bd..7d54906b7d 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/cql3/AccordInteropMultiNodeTableWalkBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/cql3/AccordInteropMultiNodeTableWalkBase.java @@ -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 diff --git a/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java b/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java index 171ccf140c..a890efed4c 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java @@ -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, diff --git a/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTokenConflictTest.java b/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTokenConflictTest.java index 090843c4b3..216b1e79c6 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTokenConflictTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTokenConflictTest.java @@ -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) diff --git a/test/distributed/org/apache/cassandra/distributed/test/cql3/StatefulASTBase.java b/test/distributed/org/apache/cassandra/distributed/test/cql3/StatefulASTBase.java index 527a7ea658..19f95f4e5a 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/cql3/StatefulASTBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/cql3/StatefulASTBase.java @@ -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 LIMIT_DISTRO = Gens.mixedDistribution(1, 1001); protected static final Gen REPAIR_TYPE_EMPTY_MODEL_DISTRO = Gens.mixedDistribution(0, 2); protected static final Gen REPAIR_TYPE_DISTRO = Gens.mixedDistribution(0, 3); + private static final ListType LONG_LIST_TYPE = ListType.getInstance(LongType.instance, false); static { @@ -206,10 +211,55 @@ public class StatefulASTBase extends TestBaseImpl }); } + protected static Property.Command 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 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 Property.Command 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 diff --git a/test/harry/main/org/apache/cassandra/harry/model/ASTSingleTableModel.java b/test/harry/main/org/apache/cassandra/harry/model/ASTSingleTableModel.java index 782994fd05..3d2bf9a74d 100644 --- a/test/harry/main/org/apache/cassandra/harry/model/ASTSingleTableModel.java +++ b/test/harry/main/org/apache/cassandra/harry/model/ASTSingleTableModel.java @@ -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 CAS_APPLIED_COLUMNS = ImmutableUniqueList.builder().add(CAS_APPLIED).build(); @@ -98,6 +115,7 @@ public class ASTSingleTableModel private final EnumSet ignoredIssues; private final TreeMap 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 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 lets(Txn txn) + { + Map 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 lets = lets(txn); + txn.output.ifPresent(select -> validate(actual, select, lets)); + List 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 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 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 lets) + private enum Who { cas, accord } + + private boolean process(Who who, Conditional condition, Map 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 key) + { + return get(factory.createRef(key)); + } + public List getByToken(Token token) { NavigableSet keys = partitions.navigableKeySet(); @@ -952,6 +1085,19 @@ public class ASTSingleTableModel return matches.stream().map(partitions::get).collect(Collectors.toList()); } + public List> 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 toRow(ImmutableUniqueList columns, ByteBuffer[][] rows) { - Set set = new HashSet<>(); + Set 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 columns(Select select) + public ImmutableUniqueList columns(Select select) { if (select.selections.isEmpty()) return factory.selectionOrder; var builder = ImmutableUniqueList.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 ""; - return symbol.type().asCQL3Type().toCQLLiteral(b); + return symbol.type().toCQLString(b); } public List asCQL() diff --git a/test/harry/main/org/apache/cassandra/harry/model/DetailedTableMetadata.java b/test/harry/main/org/apache/cassandra/harry/model/DetailedTableMetadata.java new file mode 100644 index 0000000000..51c4f8ea22 --- /dev/null +++ b/test/harry/main/org/apache/cassandra/harry/model/DetailedTableMetadata.java @@ -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 partitionColumns; + public final ImmutableUniqueList clusteringColumns; + public final ImmutableUniqueList primaryColumns; + public final ImmutableUniqueList staticColumns; + public final ImmutableUniqueList regularColumns; + public final ImmutableUniqueList selectionOrder, partitionAndStaticColumns, clusteringAndRegularColumns, regularAndStaticColumns; + + public DetailedTableMetadata(TableMetadata metadata) + { + this.metadata = metadata; + ImmutableUniqueList.Builder 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(); + } +} diff --git a/test/microbench/org/apache/cassandra/test/microbench/BatchStatementBench.java b/test/microbench/org/apache/cassandra/test/microbench/BatchStatementBench.java index 16dbbda72a..dd3e1a1d34 100644 --- a/test/microbench/org/apache/cassandra/test/microbench/BatchStatementBench.java +++ b/test/microbench/org/apache/cassandra/test/microbench/BatchStatementBench.java @@ -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); } diff --git a/test/simulator/main/org/apache/cassandra/simulator/AbstractSimulation.java b/test/simulator/main/org/apache/cassandra/simulator/AbstractSimulation.java new file mode 100644 index 0000000000..9b3fbcb2d8 --- /dev/null +++ b/test/simulator/main/org/apache/cassandra/simulator/AbstractSimulation.java @@ -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 + { + + } +} diff --git a/test/simulator/main/org/apache/cassandra/simulator/ActionSchedule.java b/test/simulator/main/org/apache/cassandra/simulator/ActionSchedule.java index 39666077ea..92ad5681e7 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/ActionSchedule.java +++ b/test/simulator/main/org/apache/cassandra/simulator/ActionSchedule.java @@ -73,7 +73,34 @@ public class ActionSchedule implements CloseableIterator, 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 { diff --git a/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java b/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java index bfc2b0576c..4db15c6c30 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java +++ b/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java @@ -141,6 +141,16 @@ public class ClusterSimulation implements AutoCloseable RunnableActionScheduler create(RandomSource random); } + public interface FutureActionSchedulerFactory + { + FutureActionScheduler create(int nodeCount, SimulatedTime time, RandomSource random); + } + + public interface PerVerbFutureActionSchedulersFactory + { + Map create(int nodeCount, SimulatedTime time, RandomSource random); + } + @SuppressWarnings("UnusedReturnValue") public static abstract class Builder { @@ -201,6 +211,8 @@ public class ClusterSimulation implements AutoCloseable protected SimulatedTime.Listener timeListener = (i1, i2) -> {}; protected LongConsumer onThreadLocalRandomCheck; protected String transactionalMode = "off"; + protected FutureActionSchedulerFactory futureActionSchedulerFactory; + protected PerVerbFutureActionSchedulersFactory perVerbFutureActionSchedulersFactory; public Builder failures(Failures failures) { @@ -506,8 +518,16 @@ public class ClusterSimulation implements AutoCloseable return this; } + public Builder 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 implements AutoCloseable new SchedulerConfig(schedulerDelayChance, schedulerDelayNanos, schedulerLongDelayNanos)); } + public Builder perVerbFutureActionSchedulers(PerVerbFutureActionSchedulersFactory factory) + { + perVerbFutureActionSchedulersFactory = factory; + return this; + } + public Map 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 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; diff --git a/test/simulator/main/org/apache/cassandra/simulator/cluster/ClusterActions.java b/test/simulator/main/org/apache/cassandra/simulator/cluster/ClusterActions.java index c8bc94d145..4d215ceefa 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/cluster/ClusterActions.java +++ b/test/simulator/main/org/apache/cassandra/simulator/cluster/ClusterActions.java @@ -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 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 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 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) { diff --git a/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThreadFactory.java b/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThreadFactory.java index 6579319395..d9d443b946 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThreadFactory.java +++ b/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThreadFactory.java @@ -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); } } diff --git a/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedTime.java b/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedTime.java index ae20d57590..df4fa90403 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedTime.java +++ b/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedTime.java @@ -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); } diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/EpochStressTest.java b/test/simulator/test/org/apache/cassandra/simulator/test/EpochStressTest.java index 8aaab3d78a..fbc976f938 100644 --- a/test/simulator/test/org/apache/cassandra/simulator/test/EpochStressTest.java +++ b/test/simulator/test/org/apache/cassandra/simulator/test/EpochStressTest.java @@ -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 diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/SimulationTestBase.java b/test/simulator/test/org/apache/cassandra/simulator/test/SimulationTestBase.java index 0c1088b6eb..aedd7b407b 100644 --- a/test/simulator/test/org/apache/cassandra/simulator/test/SimulationTestBase.java +++ b/test/simulator/test/org/apache/cassandra/simulator/test/SimulationTestBase.java @@ -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 + static abstract class BasicSimulationBuilder extends ClusterSimulation.Builder { - protected final Function init; - protected final Function test; - protected final Function teardown; + abstract S create(SimulatedSystems simulated, RunnableActionScheduler scheduler, Cluster cluster, ClusterActions.Options options); - DTestClusterSimulationBuilder(Function init, - Function test, - Function teardown) + protected void updateConfig(IInstanceConfig config) + { + + } + + public ClusterSimulation 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 + { + protected final Function init; + protected final Function test; + protected final Function teardown; + + DTestClusterSimulationBuilder(Function init, + Function test, + Function teardown) { this.init = init; this.test = test; this.teardown = teardown; } - public ClusterSimulation create(long seed) throws IOException + public ClusterSimulation 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 init, - Function test, - Function teardown, - Consumer> configure) throws IOException + public static void simulate(Function init, + Function test, + Function teardown, + Consumer> configure) throws IOException { simulate(new DTestClusterSimulationBuilder(init, test, teardown), configure); @@ -212,9 +246,49 @@ public class SimulationTestBase public static void simulate(ClusterSimulation.Builder factory, Consumer> configure) throws IOException + { + simulate(System::currentTimeMillis, factory, configure); + } + + public static void simulate(long seed, ClusterSimulation.Builder factory) throws IOException + { + simulate(() -> seed, factory, i ->{}); + } + + public static void simulate(ClusterSimulation.SimulationFactory factory) throws IOException + { + simulate(System.currentTimeMillis(), factory); + } + + public static void simulate(long seed, ClusterSimulation.SimulationFactory factory) throws IOException + { + simulate(seed, factory, b -> {}); + } + + public static void simulate(ClusterSimulation.SimulationFactory factory, Consumer> configure) throws IOException + { + simulate(System.currentTimeMillis(), factory, configure); + } + + public static void simulate(long seed, ClusterSimulation.SimulationFactory factory, Consumer> 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 void simulate(LongSupplier seedGen, + ClusterSimulation.Builder factory, + Consumer> 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, diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/SingleNodeSingleTableASTTest.java b/test/simulator/test/org/apache/cassandra/simulator/test/SingleNodeSingleTableASTTest.java new file mode 100644 index 0000000000..8f9635641e --- /dev/null +++ b/test/simulator/test/org/apache/cassandra/simulator/test/SingleNodeSingleTableASTTest.java @@ -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 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 + * + * 
{@code Caused by: java.lang.NoClassDefFoundError: Could not initialize class org.apache.cassandra.simulator.logging.RunStartDefiner}
+ */ + 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 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 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)); + } +} \ No newline at end of file diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/SingleTableASTSimulation.java b/test/simulator/test/org/apache/cassandra/simulator/test/SingleTableASTSimulation.java new file mode 100644 index 0000000000..bae0740295 --- /dev/null +++ b/test/simulator/test/org/apache/cassandra/simulator/test/SingleTableASTSimulation.java @@ -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 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 setupTable(int[] dcSizes) + { + List 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 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> uniquePartitions = Gens.lists(toGen(ASTGenerators.columnValues(model.factory.partitionColumns))) + .uniqueBestEffort() + .ofSize(rs.nextInt(1, 20)) + .next(rs); + Gen mutationGen = toGen(ASTGenerators.mutationBuilder(rs, model, uniquePartitions, i -> null).disallowEmpty().build()) + .map(mutation -> query(mutation)); + + Gen selectPartitionGen = Gens.pick(uniquePartitions) + .map(partition -> query(select(partition).build())); + Gen selectRowGen = Gens.pick(uniquePartitions).map(this::selectRow); + Gen txnGen = Generators.toGen(new ASTGenerators.ModelBasedTxnGenBuilder(rs, model, i -> null, SourceDSL.arbitrary().pick(uniquePartitions)).disallowEmpty().build()) + .map(txn -> query(txn)); + Gens.OneOfBuilder commandsBuilder = Gens.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 partition) + { + + var builder = select(partition); + List> 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> clusteringKeys = state.clusteringKeys(); + if (clusteringKeys.isEmpty()) + return null; + Clustering 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 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 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 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(); + }; + } + } +} diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/TrivialSimulationTest.java b/test/simulator/test/org/apache/cassandra/simulator/test/TrivialSimulationTest.java index 7be09a1b5c..404606c1a4 100644 --- a/test/simulator/test/org/apache/cassandra/simulator/test/TrivialSimulationTest.java +++ b/test/simulator/test/org/apache/cassandra/simulator/test/TrivialSimulationTest.java @@ -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 diff --git a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java index d44dea2115..fd1b64083a 100644 --- a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java +++ b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java @@ -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", diff --git a/test/unit/org/apache/cassandra/cql3/ast/CQLFormatterPrettyPrintTest.java b/test/unit/org/apache/cassandra/cql3/ast/CQLFormatterPrettyPrintTest.java index f14336d779..dc09a9d2c1 100644 --- a/test/unit/org/apache/cassandra/cql3/ast/CQLFormatterPrettyPrintTest.java +++ b/test/unit/org/apache/cassandra/cql3/ast/CQLFormatterPrettyPrintTest.java @@ -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), diff --git a/test/unit/org/apache/cassandra/cql3/ast/Conditional.java b/test/unit/org/apache/cassandra/cql3/ast/Conditional.java index 4fc8a20850..d2a8e2b125 100644 --- a/test/unit/org/apache/cassandra/cql3/ast/Conditional.java +++ b/test/unit/org/apache/cassandra/cql3/ast/Conditional.java @@ -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()) diff --git a/test/unit/org/apache/cassandra/cql3/ast/CreateIndexDDL.java b/test/unit/org/apache/cassandra/cql3/ast/CreateIndexDDL.java index 3cdf2eba0c..91f6167ba5 100644 --- a/test/unit/org/apache/cassandra/cql3/ast/CreateIndexDDL.java +++ b/test/unit/org/apache/cassandra/cql3/ast/CreateIndexDDL.java @@ -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 supportedQueries() + { + return indexDDL.indexer.supportedQueries(symbol.type()); + } + } + public static class CollectionReference implements ReferenceExpression { public enum Kind { FULL, KEYS, ENTRIES } diff --git a/test/unit/org/apache/cassandra/cql3/ast/FunctionCall.java b/test/unit/org/apache/cassandra/cql3/ast/FunctionCall.java index afd8bac2a3..fcba52aaa5 100644 --- a/test/unit/org/apache/cassandra/cql3/ast/FunctionCall.java +++ b/test/unit/org/apache/cassandra/cql3/ast/FunctionCall.java @@ -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() diff --git a/test/unit/org/apache/cassandra/cql3/ast/Mutation.java b/test/unit/org/apache/cassandra/cql3/ast/Mutation.java index 7c764853e5..1031fa0fd2 100644 --- a/test/unit/org/apache/cassandra/cql3/ast/Mutation.java +++ b/test/unit/org/apache/cassandra/cql3/ast/Mutation.java @@ -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> implements Conditional.EqBuilderPlus + public static abstract class TableBasedBuilder> implements Conditional.EqBuilderPlus { private final Kind kind; protected final TableMetadata metadata; protected final LinkedHashSet partitionColumns, clusteringColumns, primaryColumns, regularAndStatic, allColumns; private boolean includeKeyspace = true; - private final Set 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 + public static class InsertBuilder extends TableBasedBuilder { private final LinkedHashMap values = new LinkedHashMap<>(); private boolean ifNotExists = false; @@ -805,131 +793,121 @@ WHERE PK_column_conditions } } - public static class UpdateBuilder extends BaseBuilder implements Conditional.ConditionalBuilderPlus + public interface WithTTl> { + B ttl(Value value); + default B ttl(int value) + { + return ttl(Bind.of(value)); + } + } + + public interface WithTimestamp> + { + default B timestamp(long value) + { + return timestamp(Literal.of(value)); + } + + B timestamp(Value value); + } + + public interface UpdateBuilder> extends Conditional.ConditionalBuilder, + Conditional.EqBuilder, + WithTTl, WithTimestamp + { + 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> implements UpdateBuilder + { + private TableReference table; private @Nullable TTL ttl; private @Nullable Timestamp timestamp; private final LinkedHashMap 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 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 expressions) + public B in(ReferenceExpression ref, List 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 implements Conditional.ConditionalBuilderPlus + public static class BasicUpdateBuilder extends BaseUpdateBuilder + { + public BasicUpdateBuilder(TableReference table) + { + super(table); + } + } + + private static class PkAware + { + private final Set 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 + implements UpdateBuilder, Conditional.ConditionalBuilderPlus + { + 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 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 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 implements Conditional.ConditionalBuilderPlus { private final List 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; diff --git a/test/unit/org/apache/cassandra/cql3/ast/Select.java b/test/unit/org/apache/cassandra/cql3/ast/Select.java index 10d98dee63..ed8e0f1e72 100644 --- a/test/unit/org/apache/cassandra/cql3/ast/Select.java +++ b/test/unit/org/apache/cassandra/cql3/ast/Select.java @@ -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; diff --git a/test/unit/org/apache/cassandra/cql3/ast/Statement.java b/test/unit/org/apache/cassandra/cql3/ast/Statement.java index fd9a9ab2a9..71d85c5fb1 100644 --- a/test/unit/org/apache/cassandra/cql3/ast/Statement.java +++ b/test/unit/org/apache/cassandra/cql3/ast/Statement.java @@ -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() diff --git a/test/unit/org/apache/cassandra/cql3/ast/Txn.java b/test/unit/org/apache/cassandra/cql3/ast/Txn.java index 2e60574eac..b2ddee2f3f 100644 --- a/test/unit/org/apache/cassandra/cql3/ast/Txn.java +++ b/test/unit/org/apache/cassandra/cql3/ast/Txn.java @@ -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 mutations; + public final Conditional conditional; + public final List mutations; public If(Conditional conditional, List mutations) { diff --git a/test/unit/org/apache/cassandra/hints/HintsServiceTest.java b/test/unit/org/apache/cassandra/hints/HintsServiceTest.java index 78aa85c59e..773b30f677 100644 --- a/test/unit/org/apache/cassandra/hints/HintsServiceTest.java +++ b/test/unit/org/apache/cassandra/hints/HintsServiceTest.java @@ -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 mutateWithAccordAsync(ClusterMetadata cm, Mutation mutation, @Nullable ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime) + public IAccordResult mutateWithAccordAsync(ClusterMetadata cm, Mutation mutation, @Nullable ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime, PreserveTimestamp preserveTimestamps) { accordTxnCount.incrementAndGet(); TxnId txnId = AccordTestUtils.txnId(42, 43, 44); diff --git a/test/unit/org/apache/cassandra/service/TimestampSourceTest.java b/test/unit/org/apache/cassandra/service/TimestampSourceTest.java new file mode 100644 index 0000000000..e0bb80ea91 --- /dev/null +++ b/test/unit/org/apache/cassandra/service/TimestampSourceTest.java @@ -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)); + } +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/utils/ASTGenerators.java b/test/unit/org/apache/cassandra/utils/ASTGenerators.java index 3e1a228db1..4f502b19bb 100644 --- a/test/unit/org/apache/cassandra/utils/ASTGenerators.java +++ b/test/unit/org/apache/cassandra/utils/ASTGenerators.java @@ -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 IGNORE_ISSUES = KnownIssue.ignoreAll(); + public static final EnumSet IGNORED_ISSUES = KnownIssue.ignoreAll(); + + public static Gen> columnValues(List columns) + { + List> gens = new ArrayList<>(columns.size()); + for (int i = 0; i < columns.size(); i++) + gens.add(getTypeSupport(columns.get(i).type()).valueGen); + return rs -> { + LinkedHashMap 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 map) + { + Select.TableBasedBuilder builder = Select.builder(metadata); + for (var e : map.entrySet()) + builder.value(e.getKey(), e.getValue()); + return builder.build(); + } static Gen valueGen(Object value, AbstractType type) { @@ -87,7 +116,7 @@ public class ASTGenerators static Gen 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> partitionKeys, Gen> clusteringKeys) + public SelectGenBuilder withKeys(Gen> partitionKeys, Gen> clusteringKeys) { keyGen = rs -> { Map keys = new LinkedHashMap<>(); @@ -345,7 +380,7 @@ public class ASTGenerators { Map> 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 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 ignoreIssues = IGNORE_ISSUES; + private EnumSet 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 setBuilder = builder::set; + Conditional.EqBuilder 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 setBuilder = builder::set; + Conditional.EqBuilder 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 whereGen(Reference ref) { Gen 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> uniquePartitions, + Function indexes) + { + return mutationBuilder(IGNORED_ISSUES, rs, model, uniquePartitions, indexes); + } + + public static MutationGenBuilder mutationBuilder(EnumSet ignoredIssues, + RandomSource rs, + ASTSingleTableModel model, + List> uniquePartitions, + Function 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 indexes; + private final Gen> partitionKeyValuesGen; + private Gen bindOrLiteralGen = SourceDSL.booleans().all(); + private boolean allowEmpty = true; + + public ModelBasedTxnGenBuilder(RandomSource rs, + ASTSingleTableModel model, + Function indexes, + Gen> 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 bindOrLiteralGen) + { + this.bindOrLiteralGen = bindOrLiteralGen; + return this; + } + + private Gen mutationGen(RandomSource rs, LinkedHashMap 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 build() + { + Gen 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(); + }; + } + } }