diff --git a/doc/modules/cassandra/nav.adoc b/doc/modules/cassandra/nav.adoc index 8c28d63155..813aea24f1 100644 --- a/doc/modules/cassandra/nav.adoc +++ b/doc/modules/cassandra/nav.adoc @@ -23,6 +23,9 @@ *** xref:cassandra:architecture/guarantees.adoc[] *** xref:cassandra:architecture/messaging.adoc[] *** xref:cassandra:architecture/streaming.adoc[] +*** xref:cassandra:architecture/accord.adoc[] +**** xref:cassandra:architecture/accord-architecture.adoc[] +**** xref:cassandra:architecture/cql-on-accord.adoc[] ** xref:cassandra:developing/data-modeling/index.adoc[] *** xref:cassandra:developing/data-modeling/intro.adoc[] @@ -107,12 +110,14 @@ **** xref:cassandra:managing/operating/transientreplication.adoc[Transient replication] **** xref:cassandra:managing/operating/virtualtables.adoc[Virtual tables] **** xref:cassandra:managing/operating/password_validation.adoc[Password validation] +**** xref:cassandra:managing/operating/onboarding-to-accord.adoc[] *** xref:cassandra:managing/tools/index.adoc[Tools] **** xref:cassandra:managing/tools/cqlsh.adoc[cqlsh: the CQL shell] **** xref:cassandra:managing/tools/nodetool/nodetool.adoc[nodetool] **** xref:cassandra:managing/tools/sstable/index.adoc[SSTable tools] **** xref:cassandra:managing/tools/cassandra_stress.adoc[cassandra-stress] + ** xref:cassandra:troubleshooting/index.adoc[Troubleshooting] *** xref:cassandra:troubleshooting/finding_nodes.adoc[Finding misbehaving nodes] *** xref:cassandra:troubleshooting/reading_logs.adoc[Reading Cassandra logs] diff --git a/doc/modules/cassandra/pages/developing/accord/index.adoc b/doc/modules/cassandra/pages/architecture/accord-architecture.adoc similarity index 99% rename from doc/modules/cassandra/pages/developing/accord/index.adoc rename to doc/modules/cassandra/pages/architecture/accord-architecture.adoc index 8320b49a0a..201abd861e 100644 --- a/doc/modules/cassandra/pages/developing/accord/index.adoc +++ b/doc/modules/cassandra/pages/architecture/accord-architecture.adoc @@ -1,4 +1,4 @@ -== Accord Intro += Accord Architecture This document is intended to facilitate quick dive into Accord and Cassandra Integration code for anyone interested in the project. Readers diff --git a/doc/modules/cassandra/pages/architecture/accord.adoc b/doc/modules/cassandra/pages/architecture/accord.adoc new file mode 100644 index 0000000000..51f9f953ef --- /dev/null +++ b/doc/modules/cassandra/pages/architecture/accord.adoc @@ -0,0 +1,7 @@ += Accord + +Accord is one of the transaction protocols supported by Apache Cassandra. Accord is a separate sub-project that +is implemented as a library that is Cassandra agnostic. + +* xref:architecture/accord-architecture.adoc[] +* xref:architecture/cql-on-accord.adoc[] diff --git a/doc/modules/cassandra/pages/architecture/cql-on-accord.adoc b/doc/modules/cassandra/pages/architecture/cql-on-accord.adoc new file mode 100644 index 0000000000..91a8fc2854 --- /dev/null +++ b/doc/modules/cassandra/pages/architecture/cql-on-accord.adoc @@ -0,0 +1,612 @@ += Developers guide to CQL on Accord + +== Intro + +Accord is implemented as a library that is agnostic to the underlying +database it integrates with. It has little to no awareness of schema, +query language, messaging, threading etc. Instead it presents interfaces +for the database to implement that describe the configuration and +topology of the database, what reads and writes need to execute and what +their dependencies are, and how to actually execute reads and writes at +the configured locations. + +This guide describes how Cassandra goes about leveraging those +interfaces to implement reading and writing CQL as well as live +migrating from CQL running on Cassandra to CQL running on Accord. + +This guide doesn't cover how Accord works and doesn't cover all parts of +Accord that are implemented in Cassandra like threading, caching, +persistence, and messaging. It also isn't intended to be a user guide +and doesn't fully overlap with the xref:cassandra:managing/operating/onboarding-to-accord.adoc[user guide]. You should start with the +xref:cassandra:managing/operating/onboarding-to-accord.adoc[user guide] to get any context that may be missing here. + +== Anatomy of a transaction + +The primary way of interacting with Accord is to define a transaction +using +https://github.com/apache/cassandra-accord/blob/134df57677bbd5092994923a4dc2f15cd1d033d1/accord-core/src/main/java/accord/primitives/Txn.java#L42[Txn/Txn.InMemory] +and then asking Accord to execute the transaction. Transactions express +what they touch by declaring a set of keys or ranges that will be +read/written to. This set needs to be declared up front and can't change +during transaction execution and the transaction can be either a key +transaction or range transaction but not both. + +Range transactions are more expensive for Accord to execute as the +dependency tracking work Accord has to do is more CPU and memory +intensive and the transactions are more likely to conflict and block +execution of other transactions. + +Accord is not aware of tables only ranges and keys. Keys and ranges can +span any tables managed by Accord and the keys and ranges encode the +tables they apply to. So a range transaction covering multiple tables +would have a range per table and from Accord's perspective these are +completely different ranges. + +Transactions also declare a `Kind` which can be `Read`, `Write` +(Read/Write), `EphemeralRead`, and `ExclusiveSyncPoint`. `Read`, and +`Write` are what you would expect. `EphemeralRead` is a read that only +provides per key linearizability, but offers better performance compared +to `Read` . + +`ExclusiveSyncPoint` is transaction that can be used to establish a +happens before relationship with its dependencies without interfering +with their execution. `ExclusiveSyncPoint` is used for live migration +and repair to ensure the visibility at `ALL` of all committed Accord +transactions to non-transactional reads. + +=== Keys and Ranges + +`Keys` and `Ranges` are prefixed with `TableId` in the most significant +position to allow Accord to interact with multiple tables without +knowing anything about schema. From Accord's perspective there is just a +set of ranges that it is responsible for replicating and transacting +over, and they can be compared, sorted, and split, but beyond that they +are completely opaque. A follow on effect from this is that token ranges +(or token ring) are per table. + +`Key` is conceptually similar to `DecoratedKey` and is implemented by +`https://github.com/apache/cassandra/blob/63d3538ba7352635b7b61a205b40e035e62b8d5d/src/java/org/apache/cassandra/service/accord/api/PartitionKey.java#L43[PartitionKey]` +. `RoutingKey` is conceptually similar to `Token` and is implemented by +`https://github.com/apache/cassandra/blob/63d3538ba7352635b7b61a205b40e035e62b8d5d/src/java/org/apache/cassandra/service/accord/api/TokenKey.java#L51[TokenKey]` +. + +Accord `Range` is conceptually equivalent to Cassandra's +`Range++<++RingPosition++>++` and is implemented by +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/accord/TokenRange.java[TokenRange]`. +Accord `Range` is start exclusive and end inclusive just like +Cassandra's `Range` and we use it exclusively in that mode. There are no +other forms of inclusive/exclusive bound or range used directly by +`Accord`. Accord `Range`'s implementation suggests support for other +forms of bounds but it's not currently supported. It's theoretically +possible to use something similar to `Range++<++PartitionPosition++>++` +as the implementation of Accord's `Range` but we don't do that because +Cassandra doesn't support splitting partitions. + +To integrate Cassandra with Accord it's necessary to have a few +different versions of `TokenKey` that make it possible to describe +cluster topology and perform query routing to Accord across a range of +partitioners. A `TokenKey` can be a sentinel for a given table which +maps to `-inf` or `{plus}inf` for that table and it's possible to create +a minimum sentinel that is ++<++ `-inf` or ++>++ `{plus}inf` . +Additionally it's possible to declare a `TokenKey` that is between +`token` and either `token - 1` or `token {plus} 1` . + +Accord expects to be able to convert a `RoutingKey` to a `Range` which +is facilitated by being able to create these in between tokens without +requiring the partitioner to support increment or decrement on token. +Partition range reads also leverage these in between tokens to convert +`Range` bounds from inclusive to exclusive and vice versa to match the +inclusivity/exclusivity of the query that is being executed. + +=== Seekable, Unseekable, Routable + +The implementations of these interfaces are always prefixed with +`TableId` most of which were just discussed. + +A `Seekable` has enough information that it can be used to both route a +query and then execute it because it identifies what exactly to read and +write. An `Unseekable` is more compact (just a token) for Accord to work +with and can be used to route and schedule transaction execution. A +`Routable` could be either `Seekable` or `Unseekable` and is generally +used when you need to handle both. + +`Seekable` can be either a `Key` or an Accord `Range`. `Key` has both +routing (token) and partition key/clustering information. `Range` is +`Seekable` but its bounds are only `Routable`. `Range` is in an odd +place in terms of being `Seekable` . It's helpful because APIs can +accept `Seekable` and then handle both `Key` and `Range` domains. + +`Seekables` is the collection version of `Seekable` and can be either +`Keys` or `Ranges`. + +`Unseekable` can either a `RoutingKey` (`TokenKey`) or `Range` +(`TokenRange`) and `Unseekables` is either `RoutingKeys` or `Ranges`. +`Route` and various kinds of `Routables` exists, but are primarily used +inside Accord. + +=== Data + +`https://github.com/apache/cassandra-accord/blob/134df57677bbd5092994923a4dc2f15cd1d033d1/accord-core/src/main/java/accord/api/Data.java#L28[Data]` +is an opaque container for data that has been read during execution of a +transaction. Accord doesn't know anything about the contents and the +only required interface for `Data` is that they can be merged since +Accord will execute multiple reads at different command stores and will +need to merge the result. + +`Data` is implemented by +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/accord/txn/TxnData.java#L47[TxnData]` +which is a glorified map from a unique integer identifying each piece of +data read to `TxnDataValue` which can be either +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/accord/txn/TxnDataKeyValue.java[TxnDataKeyValue]` +or +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/accord/txn/TxnDataRangeValue.java[TxnDataRangeValue]` +. `TxnDataKeyValue` doesn't support merging because Accord only reads +from a single replica, but `TxnDataRangeValue` does because the integer +key for `TxnData` identifies the logical read in the transaction, but +the actual execution of the range read could touch an arbitrary number +of command stores covered by the range and each will produce their own +`TxnDataRangeValue` for their portion of the read. + +=== Result + +`https://github.com/apache/cassandra-accord/blob/134df57677bbd5092994923a4dc2f15cd1d033d1/accord-core/src/main/java/accord/api/Result.java[Result]` +is the interface for what is returned by `Query` and ends up being +returned as the non-error result by Accord to the coordinator of a +transaction. This is also implemented by `TxnData` for key read results +and by `TxnRangeReadResult` for range reads. + +There is also `RetryNewProtocolResult` which can be returned by +Cassandra's integration with Accord during live migration. This retry +error indicates that Accord determined the transaction's execute time is +in an epoch where Accord does not manage some or all of that data for +read or write so the transaction should be retried on whatever system +currently manages that data. + +=== Read + +`https://github.com/apache/cassandra-accord/blob/trunk/accord-core/src/main/java/accord/api/Read.java#L32[Read]` +is where a transaction defines how data should be read during execution +in order to return a result, and it will have its `read` method invoked +along with specific keys to be read at command stores. + +A `Read` has to define all the keys it will access up front and needs to +support `slice/intersecting/merge` so Accord can send only the relevant +parts of a transactions reads to the command stores that are responsible +for persisting metadata about the transaction and executing the read. + +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java[TxnRead]` +implements `Read` and is a sorted collection of +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java#L77[TxnNamedRead]`. +The name in `TxnNamedRead` refers to what is now the integer identifier +for each logical read in the transaction. `TxnNamedRead` supports both +key and range reads although not both in the same transaction. + +The name for a read is an incrementing integer encoded at planning time with the higher order bits storing +the kind of read and the lower order bits storing the index of the read. Kinds of reads include: + +* USER - let statements +* RETURNING - Returning select in `TransactionStatement` +* AUTO++_++READ - Automatically generated reads like list index set +* CAS++_++READ - Read for CAS statements + +Every read in a transaction is executed concurrently in the read stage +threadpool and the resulting `Data` (`TxnData`) is merged into a single +value. + +`TxnRead` contains a read consistency level that is not visible to +Accord that is used to declare the read consistency level that a +transaction requires. This will be discussed more later when we cover +interoperability, but if this is set then the transaction will actually +read from multiple replicas complete with short read protection and +blocking read repair. + +=== Query + +`https://github.com/apache/cassandra-accord/blob/trunk/accord-core/src/main/java/accord/api/Query.java#L31[Query]` +is the portion of the transaction definition responsible for computing +the `Result` of the transaction that will be returned at the +coordinator. It's implemented by +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/accord/txn/TxnQuery.java[TxnQuery]` +which has several different modes it can operate in. + +`Query` only has one method `compute` to compute the result and is run +on the coordinator of a transaction. There are few things `TxnQuery` is +responsible for such as validating the query is accessing data managed +by Accord generating a retry error if needed. For CAS statements it's +also responsible for checking the CAS condition and returning the +appropriate result. For range reads it's also responsible for merging +the range read results and reapplying the limit. + +`TxnQuery` also has an implementation, `UNSAFE++_++EMPTY`, used for +Accord system transactions that does no validation that Accord owns the +ranges in question. This is because from Accord's perspective it +immediately adopts all the ranges in a table when that table begins +migration to Accord, but from live migration's perspective (which Accord +can't see) there is a +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/consensus/migration/TableMigrationState.java[TableMigrationState]` +that specifies which ranges within a table are managed by Accord. + +Accord system transactions only impact Accord metadata so “they don't +exist” from the perspective of live migration and concurrent reading and +writing to data. + +=== Update + +`https://github.com/apache/cassandra-accord/blob/trunk/accord-core/src/main/java/accord/api/Update.java[Update]` +is invoked via the `apply` method on the Accord coordinator and is +responsible for taking in the `Data` from `Read` and producing the +`Write` that contains all the writes that we applied as part of +committing the transaction. + +`Update` requires support for `slice`/`intersecting`/`merge` so that +Accord only needs to distribute and persist the potentially sizable +partial or complete updates to the shards that actually need them. + +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java[TxnUpdate]` +implements `Update` and can contain completed or partial updates which +are completed when `apply` is called with the `TxnData` from `TxnRead`. +Updates that are not data dependent (blind writes) are handled +differently from non-data dependent updates. Data dependent updates are +computed at the coordinator and returned in the `TxnWrite` but non-data +dependent updates are omitted and instead are retrieved from `TxnUpdate` +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. + +`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 +write consistency level is set then Accord will do synchronous commit at +the specified consistency level. Otherwise Accord defaults to +asynchronous commit. How consistency levels are handled will be covered +in interoperability and live migration. + +=== Write + +`https://github.com/apache/cassandra-accord/blob/trunk/accord-core/src/main/java/accord/api/Write.java[Write]` +is produced by invoking `Update.apply` and is not required to be +splittable/mergeable because all writes are sent to all shards. `Write` +is implemented by +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java#L74[TxnWrite]` +which each command store will invoke via `apply` for each intersecting +key. This will cause all writes in a transaction to run concurrently on +the mutation stage. + +=== Putting it all together + +With all the components of a transaction available they can be assembled +and provided to Accord to coordinate to implement all the existing CQL +interfaces as well as the new `TransactionStatement` interface. + +See +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java#L435[TransactionStatement.createTxn]` +, +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java#L484[CQL3CasRequest.toAccordTxn]`, +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/consensus/migration/ConsensusMigrationMutationHelper.java#L236[ConsensusMigrationHelper.mutateWithAccordAsync]`, +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/StorageProxy.java#L2206[StorageProxy.readWithAccord]`, +and +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java#L219[BlockingReadRepair.repairViaAccordTransaction]` +. + +There isn't as much magic as you would think in how Accord executes +transactions when operating with exclusive access to a table. Accord is +able to mostly execute `ReadCommands` unmodified with some +accommodations for the fact that reads are strongly consistent from a +single replica so filtering can be pushed down. The majority of the work +is just making the description of things like CAS serializable so it can +be persisted by Accord for transaction recovery. + +Where things get complicated is live migrating to Accord and supporting +interoperability with non-Accord reads and writes. + +== Live migration + +=== Core challenges + +Accord and Paxos operate fundamentally different in terms of what they +perform consensus on and how the transactions are recovered. Paxos +performs consensus on the exact set of writes to apply and recovering a +transaction only requires the writes to be applied. Accord consensus is +on the transaction definition, a superset of the dependencies, and the +execution timestamp of the transaction. + +Accord needs to recompute the writes during transaction recovery which +means it may need to repeat any reads necessary to compute those writes +which means Accord needs reads to be repeatable during transaction +execution and recovery. Non-Accord writes cause non-determinism for +Accord reads. Accord also reads at `ONE` so it would miss `QUORUM` +writes. + +The big hammer we use to deal with this is to avoid ever requiring +Accord to read data that is not replicated at `ALL`. If we did it would +lead to non-deterministic transaction recovery. This isn't something +that can be addressed by having Accord read at `QUORUM` and then +performing blocking read repair because different Accord coordinators +can still witness different sets of non-Accord writes. + +Accord also defaults to asynchronous commit so when migrating away from +Accord it's not safe for Paxos and non-SERIAL reads to read committed +Accord writes + +=== Bridging the gap + +Cassandra needs to be highly available while transitioning, but +operations that propagate data at `ALL` like Cassandra's Data Repair +{plus} Paxos Repair, or Accord's repair syncs are not highly available. +Going forward these will be referred to as range barriers. + +At every point during migration there has to be some system safely +capable of executing every operation type. Highly available key barriers +solve this problem by allowing the migration of a single key at `QUORUM` +to meet the requirements for execution on the migration target system. + +A key barrier on Paxos uses the existing Paxos repair mechanism to apply +any partially committed transactions at `QUORUM` which can then be +safely read by Accord if Accord read's at `QUORUM`. A key barrier on +Accord uses Accord's sync mechanism to wait until all transactions in an +epoch that could have modified the key are applied at `QUORUM`. + +There is a system table and small in memory cache for key barriers to +avoid repeatedly performing key migrations, but the key migration is +only recorded if the coordinator is a replica to avoid the cache growing +too large. + +=== No non-SERIAL key migration + +One wrinkle is that it is not possible to do key migration for +non-SERIAL Cassandra writes because there is no metadata to check for +uncommitted operations like there is with Paxos and Accord. Non-SERIAL +writes include _all_ sources of non-SERIAL writes such as read repair, +logged batches, and hints. Accord doesn't have this issue as any data +managed by Accord always has metadata available since all operations are +routed through Accord. + +Splitting migration to Accord into two phases solves this issue +because while Accord is unable to safely read non-SERIAL writes it can +safely apply non-SERIAL writes as recovery of blind write transactions +is still deterministic in Accord. In the first phase of migration to +Accord all non-SERIAL writes are executed on Accord and synchronously +applied at the requested consistency level while a data repair (full or +incremental) runs and makes it safe for Accord to read non-SERIAL +writes. Paxos continues to execute all SERIAL writes because Accord is +unable to execute SERIAL writes since it can't read yet. + +After a data repair completes the second phase of migration to Accord +begins and all operations are executed on Accord after Paxos key +migration is run to ensure that the key being read by Accord has no +unapplied Paxos transactions. After a Paxos repair {plus} data repair +(full only) the remaining Paxos writes will be visible at `ALL` and +Accord can begin executing reads at `ONE` instead of the requested +consistency level and performing asynchronous commit and ignore the +requested commit/write consistency level. + +A quirk of incremental repair is that it flushes memtables before Paxos +repair runs and as a result it doesn't replicate at `ALL` the data that +Paxos repair propagated at `QUORUM`. Thus a full repair is required for +the second phase of migration to Accord so that the Paxos data ends up +repaired at `ALL`. It's possible, but difficult, to make the +migration three phases and track the Paxos repair independently so that +you could do Paxos repair and then use IR, but this is not currently +implemented. + +=== Supported consistency levels + +Live migration to/from Accord requires Accord to honor requested +consistency levels for read and write. Cassandra's Accord integration +only adds support for a subset of consistency levels listed in +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/accord/IAccordService.java#L75[IAccordService]` +. DC aware consistency levels are not supported along with `TWO` and +`THREE`. + +Accord will always reject unsupported consistency levels even if it will +not actually be honoring them during execution to ensure that your +application remains ready to migrate away from Accord in the future. + +In the case of `ONE` as a write/commit consistency level the commit will +silently be performed at `QUORUM` + +=== Interoperability support + +Interoperability aims to extend Accord to support reading and writing at +configurable consistency levels as well as to add support for +synchronous commit. This is facilitated by extension points in Accord +that allow injecting custom implementations for various protocol steps +via +`https://github.com/apache/cassandra-accord/blob/134df57677bbd5092994923a4dc2f15cd1d033d1/accord-core/src/main/java/accord/coordinate/CoordinationAdapter.java#L64[CoordinationAdapter]` +and +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java[AccordInteropAdapter]`. + +`AccordInteropAdapter` can inject custom versions of the `execute` and +`persist` phases and does conditionally at transaction execution time +based on the read and write consistency levels provided by `TxnRead` and +`TxnUpdate` . These consistency levels can differ from the ones +requested by the application because live migration may choose to ignore +the consistency levels when they aren't needed. + +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java[AccordInteropExecution]` +allows reading at a requested consistency level. It largely inverts +control of reading in Accord and uses Cassandra's existing Read Executor +functionality to determine what nodes to contact and what commands to +send them while providing short read protection and blocking read +repair. Read executors interface with Accord via the +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/reads/ReadCoordinator.java#L37[ReadCoordinator]` +interface which can either send a regular read message or go through +Accord to send an Accord specific read message which causes the read to +execute at the appropriate command store in the appropriate +transactional context after all dependencies have been applied. + +`ReadCoordinator` also intercepts blocking read repair during execution +of an Accord transaction and executes it through the appropriate command +store. The only legitimate way for this to occur is after Paxos key +migration the data is only propagated at `QUORUM` so it is possible that +Accord reading at `QUORUM` will find replicas to read repair. It's not +strictly necessary as we already know the data is propagated at +`QUORUM`, but the support is there. + +`ReadCoordinator` also helps apply read repair mutations via Accord in +`TransactionalMode.MIXED` and during migration by applying the read +repair mutations in Accord's execute phase instead of waiting for apply. +This is safe because read repair only proposes already committed Accord +writes or already unsafe non-SERIAL writes which aren't allowed anyways. + +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java#L48[AccordInteropPersist]` +adds support for synchronous commit and commit at a requested +consistency level. It sends `AccordInteropApply` which is a synchronous +apply message that only responds once application is complete. + +https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/consensus/TransactionalMode.java#L34[`TransactionalMode`] +defines the supported modes and +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/consensus/TransactionalMode.java#L140[commitCLForMode]` +determines the commit consistency level and +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/consensus/TransactionalMode.java#L170[readCLForMode]` +determines the read consistency level. These two methods take into +account both the requested consistency level, the table specific +migration state, the current transactional mode, and the target +transactional mode in order to decide whether to honor the requested +consistency level. + +=== Routing requests during migration + +During migration, requests race with changes to +https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/consensus/migration/TableMigrationState.java[`TableMigrationState`] +to execute and may complete or partially complete on the system they +were originally routed to. This race is resolved by allowing requests to +return a new retry on different system error response that has to be +handled by the coordinator. It's possible that a request may still +complete after receiving a retry different system error because the +target consistency level was still met. + +Migration is per table and per token range so it's possible for part of +a table to be running on Accord and part of it to be running on Paxos. +Requests can end up executing partially on Cassandra and partially on +Accord. + +==== Detecting misrouted requests + +For Paxos this is resolved in the prepare phase where a failure to meet +the required consistency level at the prepare phase means the operation +does not run on Paxos. If the prepare phase is being performed to +recover an existing transaction then it is allowed to proceed because +recovery will deterministically create the same state every time it runs +so it's safe to repeat even after key or range migration has occurred +since those would have already recovered the transaction. + +Accord determines an `executeAt` timestamp, that is deterministic even +during transaction recovery, for each transaction that includes an epoch +that corresponds to the epoch used by `TableMigrationState` and this is +used to check all the tables and keys being touched in a transaction. +`TxnQuery` then returns a retry on different system error if the any +part of the transaction is not eligible to run on Accord. + +`ColumnFamilyStore` checks every `Mutation` to see if it is marked as +allowing potential transaction conflicts. Paxos and Accord always mark +their `Mutation`s as allowing potential transaction conflicts because +they do the work to check for them directly, but non-SERIAL sources of +`Mutation`s will be subject to that check and a +`RetryOnDifferentSystemException` is thrown if the mutation is detected +to be misrouted according to the latest cluster metadata available at +the node attempting to apply that mutation. + +`ReadCommand` has a similar arrangement where each read command is +marked with whether it allows potential transaction conflicts and when +`executeLocally` is run the check is done against cluster metadata to +determine whether or not to throw `RetryOnDifferentSystemException`. +Accord always allows potential transaction conflicts on its read +commands, but Paxos does not because Paxos does not need to read data in +order to recover transactions. + +==== Splitting write requests + +For non-SERIAL writes the `Mutation` is split into the portion that will +execute on Accord and the portion that will execute on Cassandra and the +Accord portion is executed asynchronously while the Cassandra portion is +executed synchronously. If either attempt fails due to misrouting the +write is re-split with updated cluster metadata and retried without +raising an error. + +Logged batches are currently always written to the system table and then +split for execution, and if part of the batch fails then batchlog replay +will replay the entire batch and re-split it in the process. Batchlog +replay only makes a single attempt to replay before converting the batch +contents to hints. If part of the batch was routed to Accord then there +is no node to hint so there is a fake node that a hint is written to and +when that hint is dispatched it will be split and then executed +appropriately. In https://issues.apache.org/jira/browse/CASSANDRA-20588[CASSANDRA-20588] this needs to be simplified to writing the +entire batch through Accord if any part of it should be written through +Accord because it also addresses an atomicity issue with single token +batches which can be torn when part is applied through Accord and part +is applied through Cassandra. + +Hints can be for multiple tables some of which may be Accord and some +non-Accord so splitting occurs. It's also possible a hint will be for an +operation that was sent to Accord (not a real node) via the batchlog and +it's possible that splitting discovers the hint now needs to be executed +without Accord. In that scenario the hint is converted to a hint for +every replica. This conversion can only occur once so the write +amplification is bounded. + +Splitting of mutations is done in +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/consensus/migration/ConsensusMigrationMutationHelper.java#L219[ConsensusMigrationMutationHelper]` +with the retry loop being implemented at each caller (batch mutation, +mutation, batch log, hints). + +Paxos has a retry loop but does not do any splitting because Paxos only +supports a single key. + +==== Partition range reads + +Partition range reads are managed by +`https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/reads/range/RangeCommandIterator.java#L75[RangeCommandIterator]` +which continues to split range reads using the existing algorithm that +is agnostic as to how the range command will be executed. Each generated +range read +https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/reads/range/RangeCommandIterator.java#L247[is +then split on the boundaries of which system is responsible for reading +that range] and that is wrapped in a +https://github.com/apache/cassandra/blob/122f5300855d56131948575f80ce0594547c9040/src/java/org/apache/cassandra/service/reads/range/RangeCommandIterator.java#L378[retrying +iterator] which repeats the splitting if any part of the range read ends +up routed to the wrong system. + +Range reads do not execute any key barriers and when migrating away from +Accord you will see weaker consistency compared to Paxos because Accord +does not necessarily honor commit consistency levels and does +asynchronous commit. As things currently stand it's uncertain the key +barriers would run fast enough to avoid timing out range read requests +so they are not done. + +Range reads also consume more memory when executed on Accord when a +limit is used. A single range read command is split into intersecting +command store number of range read commands that execute concurrently +and each one can return up to the limit number of results before they +are merged at the coordinator and the limit is re-applied. This could be +improved by applying the limit again before serializing or by executing +the reads serially at command stores until the limit is met. + +=== Transactional modes + +Transactional modes are set per table and define how Accord, Paxos, and +non-SERIAL operations will execute. The three supported modes are +`FULL`, `MIXED++_++READS`, and `OFF`. + +`FULL` routes all reads and writes through Accord once migration is +complete and allows Accord to ignore read and write consistency levels. +This allows Accord to perform asynchronous commit reducing the number of +WAN roundtrips from 2 to 1. + +`MIXED++_++READS` routes all writes through Accord once migration is +complete, but allows non-SERIAL reads to safely execute outside of +Accord and still read Accord writes because Accord will honor the +provided commit consistency level. This means Accord will need to +perform synchronous commit requiring an 1 extra WAN roundtrips for 2 +total. + +`OFF` is the default where everything runs either on Paxos if it is +`SERIAL` or on the usual eventually consistent paths for everything +else. + +Other modes exist for testing purposes and are disabled by default +unless unlocked via system property. diff --git a/doc/modules/cassandra/pages/architecture/index.adoc b/doc/modules/cassandra/pages/architecture/index.adoc index 9e674d95a2..893c2f7807 100644 --- a/doc/modules/cassandra/pages/architecture/index.adoc +++ b/doc/modules/cassandra/pages/architecture/index.adoc @@ -7,3 +7,4 @@ This section describes the general architecture of Apache Cassandra. * xref:architecture/storage-engine.adoc[Storage Engine] * xref:architecture/guarantees.adoc[Guarantees] * xref:architecture/snitch.adoc[Snitches] +* xref:architecture/accord.adoc[Accord] diff --git a/doc/modules/cassandra/pages/managing/operating/index.adoc b/doc/modules/cassandra/pages/managing/operating/index.adoc index 2fb9859431..8068bd3dc4 100644 --- a/doc/modules/cassandra/pages/managing/operating/index.adoc +++ b/doc/modules/cassandra/pages/managing/operating/index.adoc @@ -20,3 +20,4 @@ * xref:cassandra:managing/operating/transientreplication.adoc[Transient replication] * xref:cassandra:managing/operating/virtualtables.adoc[Virtual tables] * xref:cassandra:managing/operating/password_validation.adoc[Password validation] +* xref:cassandra:managing/operating/onboarding-to-accord.adoc[] diff --git a/doc/modules/cassandra/pages/managing/operating/onboarding-to-accord.adoc b/doc/modules/cassandra/pages/managing/operating/onboarding-to-accord.adoc new file mode 100644 index 0000000000..17d4515000 --- /dev/null +++ b/doc/modules/cassandra/pages/managing/operating/onboarding-to-accord.adoc @@ -0,0 +1,354 @@ += Onboarding to Accord + +== Intro + +Accord supports all existing CQL and can be enabled on a per table and +per token range within that table basis. Enabling Accord on existing tables requires a +migration process that can be done on this same per table and per range +basis that safely transitions data from being managed by Cassandra +{plus} Paxos to Cassandra {plus} Accord without downtime. + +A migration is required because Accord can't safely read data written by +non-SERIAL writes. Accord requires deterministic reads in order to have +deterministic transaction recovery and non-SERIAL writes can't be read +deterministically while still being highly available. + +This guide describes how to enable Accord and what differences to expect +when migrating your existing CQL workload to Accord. + +This guide does not cover the new transaction syntax. + +== Configuration + +=== YAML + +You need to set `accord.enabled` to true for Accord to be initialized at +startup. + +`accord.default++_++transactional++_++mode` allows you to set a default +transactional mode for newly created tables which will be used in create +table statements when no `transactional++_++mode` is specified. This +prevents accidentally creating non-Accord tables that will need +migration to Accord. + +`accord.range++_++migration` configures the behavior of altering the +`transactional++_++mode` of a table. When set to `auto` the entire ring +will be marked as migrating when the `transactional++_++mode` of a table +is altered. When set to `explicit` no ranges will be marked as migrating +when the `transactional++_++mode` of a table is altered. + +=== Table parameters + +`transactional++_++mode` can be set when a table is created +`CREATE TABLE foo WITH transactional++_++mode = ‘full'` or it can be set +by altering an existing table +`ALTER TABLE foo WITH transactional++_++mode = ‘full'`. +`transactional++_++mode` designates the target or intended transaction +system for the table and for a newly created table this will be the +transaction system that is used, but for existing tables that are being +altered the table will still need to be migrated to the target system. + +`transactional++_++mode` can be set to `full`, `mixed++_++reads`, and +`off`. `off` means that Paxos will be used and transaction statements +will be rejected. `full` means that all reads and writes will execute on +Accord. `mixed++_++reads` means that all writes will execute on Accord +along with `SERIAL` reads/writes, but non-SERIAL reads/writes will +execute on the existing eventually consistent path. Applying the +mutations for blocking read repair will always be done through Accord in +`full` in and `mixed++_++reads`. + +`transactional++_++migration++_++from` indicates whether a migration is +currently in progress although it does not indicate which ranges are +actively being migrated. This is set automatically when you create a +table or alter `transactional++_++mode` and should not be set manually. +It's possible to manually set `transactional++_++migration++_++from` to +force the completion of migration without actually running the necessary +migration steps. + +`transactional++_++migration++_++from` can be set to `none`, `off`, +`full`, and `mixed++_++reads`. `off`, `full`, and `mixed++_++reads` +correspond to the `transactional++_++mode` being migrated away from and +`none` indicates that no migration is in progress either because the +migration has completed or because the table was created with its +current `transactional++_++mode`. + +=== mixed++_++reads vs full + +When Accord is running with `transactional++_++mode` `full` it will be +able to perform asynchronous commit saving a WAN roundtrip. +`mixed++_++reads` allows non-SERIAL reads to continue to execute using +the original eventually consistent read path. `mixed++_++reads`, unlikes +`full`, always requires Accord to always synchronously commit at the +requested consistency level in order to make acknowledged Accord writes +visible to non-SERIAL reads. + +There is no `transactional++_++mode` that allows non-SERIAL writes +because they break Accord's transaction recovery resulting in +transactions appearing to have different outcomes at different nodes. + +== Accord repair + +Repair can now include an optional Accord repair that `nodetool repair` +will enable by default like Paxos repair. This repair doesn't actually +synchronize any data it just runs a transaction that checks that Accord +has resolved the state of all transactions in the repaired range up to +the point the transaction was created and that the transactions are +applied at `ALL`. + +Accord is normally doing this in the background anyways this just +ensures that it has occurred at `ALL` and hasn't experienced any delays. + +== Migration to Accord + +Migrating an existing table to run on Accord starts by altering the +table: + +.... +ALTER TABLE foo WITH transactional_mode = 'full' +.... + +After the table is altered it is required to run +`nodetool consensus++_++admin begin-migration` on ranges in the table +unless `accord.range++_++migration=auto`. + +When a range is initially marked migrating to Accord all non-SERIAL +writes will execute on Accord while `SERIAL` writes will continue to +execute on Paxos. non-SERIAL writes include regular writes, logged and +unlogged batches, hints, and read repair. Accord will perform +synchronous commit the specified consistency level requiring 2x WAN RTT. + +Tables that are migrating or are partially migrated to Accord (or back to Paxos) can be listed using +`nodetool consensus_admin list` or the sytem table `system_accord_debug.migration_state`. + +Migration to Accord consists of two phases with the first phase starting +when a range is marked migrating, and the second phase starting after a +full or incremental data repair, and then the migration completing after +a second repair which must be a full data repair {plus} Paxos repair. +While marking the range as migrating can be done automatically with +`accord.range++_++migration=auto`, there is not automation for +triggering the repairs. If you regularly run compatible repairs then the +migration will eventually complete, but if you don't run them or want +the migration to complete sooner then you will need to either trigger +them manually or invoke `nodetool consensus++_++admin finish-migration` +to trigger them. + +Any repair that is compatible will drive migration forward whether it +only covers part of the migrating range or whether is started via +`nodetool consensus++_++admin finish-migration` or some other external +process that initiates repair. Force repair with down nodes will not be +eligible to drive any type or phase of migration forward. Force repair +with all nodes up will still work. + +=== First phase + +In the first phase of migration Accord is unable to safely read +non-SERIAL writes so Paxos continues to be used for `SERIAL` operations +and Accord executes all writes and synchronously commits at the +requested consistency level in order to allow Paxos to safely read +Accord writes. Accord's read and write metrics are all counted towards the existing `Read` and `Write` scope +along with the eventually consistent operations, but you should also start to see writes also being counted in the `AccordWrite` scope. + +A data repair either incremental or full replicates all non-SERIAL +writes at `ALL` making it safe for Accord to read non-SERIAL writes that +occurred before the migration started. non-SERIAL writes that occurred +after the migration started were executed through Accord so Accord can +safely read them. + +=== Second phase + +In the second phase all reads and writes execute through Accord +(assuming `transactional++_++mode="full"`). Before an operation can execute on +Accord it is necessary to run a Paxos key repair in order to ensure that +any uncommitted Paxos transactions are committed and this check will +take at least one extra WAN RTT. Additionally Accord has to read at `QUORUM` +(where it would normally only read from a single replica in `transactional++_++mode="full"` and migration completed) because +Paxos writes are only visible at `QUORUM`. + +All reads and CAS operations in the range should start showing up in the +Accord metrics and not the existing metrics. + +Once a key has been repaired, the repaired state of the key is stored in +a small in-memory cache and system table so that it doesn't need to be +repaired again. This information is only stored at replicas of the key +so if the coordinator is not a replica it will not know that it can skip +repairing the key. Use token aware routing to avoid redundant key +repairs. + +A full repair {plus} Paxos repair is necessary to complete the second +phase of migration to Accord. An incremental repair can't currently be +used because incremental repair doesn't include the transactions that +are repaired by Paxos repair because it selects the data to include in +the repair before running the Paxos repair. + +== Migration from Accord + +Migration from Accord to Paxos occurs in a single phase and begins by +altering the table's `transactional++_++mode` to `off` and then +optionally marking ranges as migrating as discussed above. + +Once a range is marked migrating all operations in the migrating range +will stop executing on Accord. Before each operation occurs they will +have to run an Accord key repair similar to the Paxos key repair to +ensure Accord transactions for that key have committed at `QUORUM`. + +An Accord repair needs to be run on the migrating range, triggered +manually or via `nodetool finish-migration`, and once that completes +non-SERIAL operations will run using the usual eventually consistent +path and `SERIAL` operations will execute on Paxos. + +== Migration commands + +All the `nodetool` migration commands are based on new +`StorageServiceMBean` JMX methods. These methods are +`migrateConsensusProtocol`, `finishConsensusMigration`, +`listConsensusMigrations`, `getAccordManagedKeyspaces`, and +`getAccordManagedTables` and can be used by external management tools to +manage consensus migration. The existing methods for starting repairs +can also be used to start the repairs that are needed to complete +migration. + +=== nodetool consensus++_++admin list + +Invoking `nodetool` with +`consensus++_++admin list ++[<++keyspace++>++ ++<++tables++>++...++]++` +will connect to the specified node and retrieve that nodes view of what +tables are currently being migrated from transactional cluster metadata. +Tables that are not being migrated are not listed. + +The results can be printed out in several different formats using the +`format` parameter which supports `json`, `minified-json`, `yaml`, and +`minified-yaml`. + +=== nodetool consensus++_++admin begin-migration + +Invoking `nodetool` with +`consensus++_++admin begin-migration ++[<++keyspace++>++ ++<++tables++>++...++]++` +can be used to mark ranges on a table as migrating. This can only be +done after the migration has been started by altering the tables. +Marking ranges as migrating is a lightweight operation and does not +trigger the repairs that will finish the migration. + +The range to mark migrating needs to be explicitly +provided otherwise the entire ring will be marked migrating for the +specified keyspace and tables. If the entire range is marked migrating it is +only necessary to invoke `begin-migration` on one node. + +This is only needed if +`accord.default++_++transactional++_++mode=explicit` is set in +`cassandra.yaml` otherwise all the ranges will already have been marked +migrating when the alter occurred. + +Ranges that are migrating will require at least an extra WAN roundtrip +for each request that touches a migrating range because both transaction +systems may need to be used to execute the request. + +=== nodetool consensus++_++admin finish-migration + +Invoking `nodetool` with +`consensus++_++admin finish-migration ++[<++keyspace++>++ ++<++tables++>++...` +will run the repairs needed to complete the migration for the specified +ranges. If no range is specified it will default to the primary range of +the node that `nodetool` is connecting to so you can call it once on +every node to complete migration. + +When migrating from Paxos to Accord it will run an incremental data +repair and then a full data repair {plus} Paxos repair. When migrating +from Accord to Paxos it will run an Accord repair. + +== Supported consistency levels + +Migration requires support for read and write consistency levels because +Accord ends up being required to read Paxos writes at `QUORUM` and +Accord needs to execute non-SERIAL writes while Paxos is still being +used for `SERIAL` writes and thus needs to perform synchronous commit at +the requested consistency level. + +Once migration is complete the read and write consistency levels will be +ignored with transactional mode `full` . With transactional mode +`mixed++_++reads` Accord will continue to do synchronous commit and +honor the requested commit/write consistency level. + +Accord will always reject any requests to execute at unsupported +consistency levels to ensure that migration to/from Accord is always +possible. + +Supported read consistency levels are `ONE`, `QUORUM`, `SERIAL`, and +`ALL`. Supported write consistency levels are `ANY`, `ONE`, `QUORUM`, +`SERIAL`, and `ALL`. `LOCAL`, `TWO`, and `THREE` are not supported. +`ANY` is executed as an asynchronous commit similar to Paxos. + +== 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. + +`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. + +Paging runs a separate transaction per page and does not produce a +linearizable result. + +Partition range reads are split into multiple transactions during +execution and will not produce a strict serializable result. +Additionally during migration there are no barriers/repairs executed +before partition range reads. When migrating from Accord to Paxos the +effective commit CL for Accord writes as viewed from partition range +reads will be `ANY`. Adding barriers/repairs before partition range +reads would cause them to time out so they are not done. + +== Batchlog and hints + +Pre-existing batchlog entries and hints will be processed during and +after migration until they are completed. If they need to be executed +through Accord they will be routed through Accord automatically. + +Logged batches that only touch Accord data will not be written to the +batch log because that functionality is redundant with Accord. Batches +that touch Accord and non-Accord data continue to use the batch log. +Before release this is likely to change so that a batch that touches +Accord data will be written entirely via Accord including both the +Accord and non-Accord data. + +Hints are not written for Accord writes although the batch log may +result in new hints because batch log entries are converted to hints +after the first retry. + +== Operations spanning Accord/non-Accord data + +Various operations can access both Accord and non-Accord managed data. +These are transparently split into parts that execute on Accord and +parts that execute outside of Accord and the results are merged. If the +splitting process races with migration then the operations is re-split +and retried without surfacing an error to the client. + +== Partition range read with LIMIT performance + +Partition range reads with a limit use more memory and CPU at the nodes +being read from and at the coordinator. Accord splits the ranges owned +by each node into smaller subranges and each subrange is owned by a +command store. The partition range read will execute at every +intersecting command store on a node and each will return `LIMIT N` +results which are sent back to the coordinator. The coordinator then +merges them and re-applies the limit. + +The additional memory and CPU will be amplified proportional to the +number of command stores which defaults to +`DatabaseDescriptor.getAvailableProcessors()`. + +== Metrics + +Accord's read and write metrics are counted under the existing `Read` and `Write` scope along with eventually consistent +operations. To see Accord specific metrics you can look at the `AccordRead` and `AccordWrite` scope. `CASRead` and `CASWrite` will not track +CAS or `SERIAL` read operations that end up running on Accord and they will instead show up in `AccordRead`/`AccordWrite` and `Read`/`Write`. + +If a single request ends up running on both systems due to misrouting it +will show up as multiple requests. Misrouted requests are counted under the `RetryDifferentSystem` meter and will show +up in `AccordRead` and `AccordWrite` if Accord was the system the request was misrouted to as well as `Read` and `Write`. +If the request was misrouted to non-Accord code then it will show up under `Read` and `Write` metrics or `CASRead` and `CASWrite` metrics. + +Hints can be misrouted and this is tracked in `HintsServiceMetrics` under the `HintsRetryDifferentSystem` meter. + +Partition range reads can also potentially generate additional Accord +transactions depending on how the reads end up having to be split due to intersection with migrating ranges.