Glossary

Atomicity, Consistency, Isolation, Durability: the safety guarantees a transaction is supposed to provide.

The term is more marketing than definition: implementations differ widely, and 'isolation' in particular rarely means serializable by default.

A consumer acknowledges each message after processing; unacknowledged messages are redelivered if the consumer dies.

Redelivery prevents loss, but combined with load balancing it reorders messages: a redelivered message can reach another consumer after later ones were already processed.

Run every transaction one at a time on a single thread, with data in memory and transactions submitted as stored procedures.

No concurrency control at all, but throughput is capped at one CPU core per partition, and cross-partition transactions are much slower.

Choosing timeouts from the measured response-time distribution and its jitter instead of a constant.

Systems that observe round-trip times can raise their suspicion level gradually, which balances detection delay against false positives as conditions change.

A data file that is only ever added to at the end; updates and deletes are new records, never in-place edits.

Appending is the cheapest write a disk can do and makes crash recovery simple, but without an index every read has to scan the whole log.

Expressing derived datasets as functions from input streams to output state, run by stream processors.

Instead of application code reading and writing a shared mutable database, each view is a deterministic function of the log. Adding a view means wiring a new consumer, not changing the writers.

A network that gives no guarantee when, or whether, a packet will arrive.

When a request gets no response, it could have been lost, queued, the remote crashed or paused, or the response was lost. The sender cannot tell these apart, and a retry may execute the operation twice.

The leader acknowledges a write as soon as it is stored locally and sends it to followers in the background.

Writes are fast and keep working when followers fail. However, writes that were acknowledged but not yet replicated are lost if the leader dies.

Ensuring that a transaction spanning several nodes either commits on all of them or aborts on all of them.

Committing independently on each node risks some committing while others abort, violating atomicity. Once committed, a transaction cannot be retracted, so the decision must be made together.

If a transaction can't finish, every write it made is discarded; it's all or nothing.

It has nothing to do with concurrency. Its value is that after a fault the application can safely retry instead of guessing which writes landed.

A Hadoop-born binary format whose encoding contains no tags and no types, only values concatenated in schema order.

It is the most compact (32 bytes), and because there are no tags, schemas can be generated from database tables and evolve freely. The cost: you cannot decode a single byte without knowing exactly which schema wrote it.

A balanced tree of fixed-size pages (typically 4 KB), each holding keys and references to child pages, overwritten in place.

With branching factor b and n keys the depth is O(log_b n), so a four-level tree of 4 KB pages with b = 500 holds 256 TB. Every key lives in exactly one place.

Blocking the producer when the consumer cannot keep up, instead of dropping messages or buffering without bound.

The three options for an overloaded consumer are drop, buffer, or block. Buffering only postpones the problem and raises latency; dropping loses data; backpressure slows the whole pipeline to the speed of its slowest part.

Newer code can read data that was written by older code.

Usually easy because the new code's author knows the old format. Rule: any field you add must be optional or have a default, otherwise new code fails on old records.

Binary encodings of the JSON data model, such as MessagePack, BSON or CBOR.

They shave some bytes and parsing cost, but because there is no schema every record still carries all its field names, so the savings are modest (81 → 66 bytes in the book's example).

One bitmap per distinct value of a column, then compressed by storing run lengths of zeros and ones.

Columns with few distinct values compress dramatically, especially when sorted. Filters become bitwise AND/OR on compressed bitmaps, executed in vectorized fashion.

A compact bit array that can say a key is definitely absent or possibly present.

LSM-trees keep one per SSTable so lookups for missing keys skip almost every table instead of reading them all from disk.

A map-side join: the small input is loaded entirely into an in-memory hash table in every mapper, which then streams its block of the large input and looks keys up.

No reducers, no sort and no shuffle, so it is much faster. It only works when one side is small enough to fit in each mapper's memory.

Nodes that behave arbitrarily: they send corrupted or conflicting messages, or lie on purpose.

Tolerating f such nodes needs n ≥ 3f + 1, compared with 2f + 1 for crash faults. It matters for aerospace and blockchains. In a datacenter, nodes are usually assumed honest, but checksums and input validation still guard against weak forms of lying.

During a network partition, a system must choose between linearizability and availability.

Partitions are not optional, so CAP really says: when one happens, either some clients get errors (stay consistent) or everyone gets answers that may be stale (stay available). It is a narrow theorem, more slogan than design tool.

Operations that are causally related are seen by everyone in the same order; concurrent ones may be seen in any order.

Causality gives a partial order, linearizability a total one. Causal consistency is the strongest model that does not slow down with network delay and stays available during partitions.

Observing every write to a database and exposing it as a stream that other systems can apply in the same order.

It makes the database the leader and derived systems followers of its replication log, so they converge on the same state. Initial state comes from a snapshot or a compacted log.

Treating a clock reading as a range [earliest, latest] with an explicit uncertainty bound.

If two intervals don't overlap, the order of events is known. Spanner waits out the uncertainty before committing, which is why Google puts GPS receivers and atomic clocks in every datacenter to keep it around 7 ms.

Quartz clocks run fast or slow depending on temperature; NTP corrects them over the network, only as accurately as the network delay allows.

Google assumes 200 ppm of drift: about 6 ms between syncs every 30 s, or 17 s a day. A firewalled NTP server, a misconfiguration or a VM pause can leave a clock badly wrong without anyone noticing.

Storing all values of each column together in separate files instead of each row together.

An analytic query that touches 4 of 100 columns reads only those 4 columns' data, cutting disk bandwidth by an order of magnitude or more.

A command is a request that may still be rejected; once validated it becomes an event, an immutable fact.

Validation must happen synchronously before the event is written. Consumers of the event cannot reject it, only compensate with further events.

Write a new value only if the current value still equals what you read; otherwise retry.

Avoids locks, but under contention most of the work is spent on retries. Unsafe if the 'compare' reads from an old snapshot.

An atomic operation that sets a value only if it currently equals an expected value, and reports whether it succeeded.

With a linearizable register plus CAS you can implement locks, leader election and uniqueness constraints; whether a CAS may succeed depends on the single global order of all writes.

Fixing a constraint violation after the fact: refund, reassign, or apologize to the customer.

Many businesses already tolerate brief violations (overbooked flights, oversold stock) because a cheap apology beats the cost of strict coordination. The question is how often it happens and how expensive each apology is.

A key with several columns where only the first is hashed to pick the partition and the rest are kept sorted within it.

A middle ground between hash and range: e.g. (user_id, timestamp) spreads users across partitions but still allows an efficient range scan over one user's updates.

A way to make every replica reach the same final value after concurrent writes: last write wins, replica priority, merging values, or keeping all versions for the application or user to resolve.

If each replica simply applies writes in the order they arrive, replicas end up permanently different. Any automatic rule has to be deterministic so that all replicas reach the same result.

Getting several nodes to agree on a value, with uniform agreement, integrity, validity and termination.

Leader election, atomic commit, uniqueness, linearizable CAS and total order broadcast all reduce to it. Unlike 2PC, fault-tolerant consensus keeps making progress as long as a majority is up.

Application-defined invariants (e.g. credits equal debits) hold before and after every transaction.

It's a property of the application, not the database: the database can't know your invariants, it can only give you atomicity and isolation to preserve them.

Place randomly chosen tokens on a hash ring; a key belongs to the next token clockwise, so adding or removing a node moves only nearby keys.

It was designed for distributed caches. The book notes the name is confusing and the scheme is rarely used as originally described; in databases, think 'hash partitioning with random boundaries'.

Anyone reading a sequence of writes sees them in the order they happened.

In a partitioned database, partitions replicate independently, so an answer can show up before its question. Putting causally related writes in the same partition, or tracking their dependencies, fixes it.

Each consumer group records how far it has read in each partition as a single offset number.

Offsets make acknowledgment cheap and replay trivial: rewind the offset to reprocess. A consumer that lags further behind than the retention window loses the messages that were deleted.

A small, fault-tolerant, linearizable key-value store that applications use for locks, leader election, membership and configuration.

It outsources consensus: few apps should implement Raft themselves. It offers linearizable atomic operations, total ordering (zxid as fencing tokens), failure detection via sessions, and change notifications.

A separate, consistent store of the partition-to-node mapping that nodes register with and routers subscribe to.

It gives one authoritative assignment and notifies subscribers when it changes, so routing can catch up. Many systems rely on it instead of building their own consensus.

Systems that preserve integrity using logs, deterministic derivation and end-to-end ids instead of synchronous cross-node coordination.

They keep working across datacenters and network faults, with better performance, while giving up only timeliness. Synchronous coordination can be reserved for the few places where it truly cannot be avoided.

Linearizable reads and writes have response times at least proportional to network delay.

Most systems that drop linearizability (even multicore CPU caches) do it for speed, not for partition tolerance. On a WAN with variable delays, every linearizable operation pays the round trip.

Chaining each log entry to the hash of the previous one (or using Merkle trees) so tampering is detectable.

Rewriting an old event breaks every hash after it. The same idea underlies certificate transparency and blockchains — a way to make the log's integrity verifiable by others.

Neo4j's declarative graph query language, built around arrow patterns like (a)-[:WITHIN*0..]->(b).

Variable-length paths make walking a hierarchy of unknown depth a single line. The SQL equivalent needs a recursive common table expression.

A database contains values written over years by many versions of the application, while only the latest code is deployed.

Rewriting (migrating) everything to the new schema is expensive, so most databases evolve lazily: old rows stay as they are and missing columns are filled in with defaults when read.

A separate read-only database for analysts, loaded from OLTP systems by Extract–Transform–Load jobs or streams.

Analysts can run expensive queries without hurting the transactional systems, on a schema and storage layout built for analytics.

An engine that runs a whole workflow as one job: a graph of operators that hand data to each other directly, sorting only where needed.

Operators can start as soon as input arrives, intermediate data stays in memory or on local disk, and there is no repeated task start-up. That makes it much faster than a chain of MapReduce jobs.

A logic language of facts and rules; rules derive new facts and can be recursive.

Older than SQL and the foundation of later graph query languages. Rules compose like functions, so you build up complex queries a piece at a time.

Transactions each waiting for a lock the other holds, forming a cycle in the wait-for graph.

The database detects the cycle and aborts one transaction, which the application must retry. Under 2PL this happens often.

You state the pattern of data you want; the database decides how to find it (SQL, CSS, Cypher, SPARQL).

Hides implementation details, so the database can add indexes, reorder work or run in parallel without your query changing.

Killing processes or machines on purpose to make sure fault-tolerance machinery really works.

Error-handling code that never runs is probably broken. Exercising it continuously builds confidence that real faults will be handled.

Caches, indexes, materialized views and models computed from the system of record by some transformation.

Derived data is redundant by design: if lost or wrong, you can recompute it. That property is what makes it safe to have many specialized stores for one dataset.

A transaction sees another transaction's writes before they've committed.

The other transaction may still abort, so you'd act on data that never officially existed. Prevented by read committed and above.

A transaction overwrites a value that another uncommitted transaction has written.

Interleaved writes across rows (the car sold to Bob but invoiced to Alice). Almost every database prevents it with row-level write locks held until commit.

A programming model where actors with local state communicate only by asynchronous messages, transparently across nodes.

Message loss is assumed even locally, so going distributed changes less than with RPC, but rolling upgrades still need the message encoding to be forward and backward compatible.

A shared-nothing filesystem: a daemon on each machine stores blocks, a central NameNode tracks where they are, and blocks are replicated.

It is the stdin/stdout of MapReduce. Jobs read input and write output as files in it, and the scheduler runs each map task on a machine that already holds that block (data locality).

Self-contained documents (usually JSON) that nest their one-to-many children inside.

A good fit when data is a tree that gets loaded all at once, such as a résumé. Weak support for joins makes highly interconnected data awkward.

Each partition maintains a secondary index covering only the documents stored in that partition.

Writes stay on one partition and the index is updated synchronously. But a query by secondary key must be sent to every partition and the results merged.

Application code writing the same change to several systems (database, cache, search index) itself.

Two concurrent clients can have their writes arrive in different orders at each system, leaving them permanently inconsistent, and one write can fail while the other succeeds.

Committed data is not lost, even after a crash: written to non-volatile storage (with a write-ahead log) or replicated before acknowledging.

Perfect durability doesn't exist (disks fail, fsync lies, correlated outages); it's a matter of reducing risk, usually by combining disks and replication.

Key-range partitions split in two when they exceed a size threshold and merge with a neighbour when they shrink, like a B-tree's top level.

The number of partitions adapts to the data volume. An empty database starts with one partition, so all writes hit one node until the first split; pre-splitting helps if you know the key distribution.

Adding computing resources automatically when load increases.

Useful when load is unpredictable, but autoscalers react with a delay. Manually scaled systems are simpler and have fewer operational surprises.

Translating an in-memory data structure into a self-contained sequence of bytes, and back (decoding).

Pointers and objects only make sense inside one process. Anything written to disk or sent over a network has to be encoded, and the format you choose decides size, speed and how painfully your schema can change.

A znode tied to the client session that created it; when heartbeats stop and the session times out, it is deleted automatically.

It turns failure detection into data: a lock or membership entry disappears when its owner dies. But a paused client may not know its session expired, so the resources it guards must check fencing tokens.

A monotonically increasing number attached to each leadership; a higher epoch always wins.

A leader cannot trust that it is still leader. Before deciding anything it needs a quorum of acknowledgments in its epoch, so a deposed leader from an older epoch can never get a value committed.

A small, self-contained, immutable record that something happened at some point in time.

Events are generated once by a producer and processed by many consumers. Thinking of data as a sequence of events, rather than as mutable state, is the foundation for everything else in the chapter.

Storing every change to application state as an immutable event in an append-only log, and deriving current state by replaying it.

Events capture intent at the application level, history is never lost, and new read-optimized views can be derived from the same events. Snapshots keep replay fast.

Event time is when the event occurred (by the device's clock); processing time is when the stream processor sees it.

Queueing, network faults, and restarts delay events, sometimes by minutes. Windowing by processing time produces artifacts such as spikes after a backlog, so rates should be computed by event time.

If writes stop, all replicas eventually return the same value; until then, reads may return anything older.

It is a very weak guarantee: it says nothing about when replicas converge, and it makes reads after writes unpredictable. Stronger models cost performance, which is why many systems settle for it.

Making it easy to adapt the system to new requirements (also called extensibility or plasticity).

Requirements always change. How easily you can change a data system depends a lot on how simple its abstractions are.

When the leader fails, the system detects it (usually with a timeout), promotes the most up-to-date follower, and redirects clients and other followers to it.

If the timeout is too short, the system fails over when nothing is wrong. If it is too long, recovery is slow. With async replication, writes the old leader had not replicated are thrown away.

Building home timelines either by merging followees' tweets at read time, or by pushing each tweet into every follower's cached timeline at write time.

Pushing on write makes reads cheap but turns a celebrity's tweet into millions of writes. Twitter ended up with a hybrid: fan out most users' tweets, and merge celebrities' tweets in at read time.

A fault is one component deviating from its spec; a failure is the system as a whole no longer serving the user.

You can't drive faults to zero, so fault-tolerant design aims to stop faults from becoming failures. Redundancy masks a dead disk; nothing masks a fault that hits every replica.

Algorithms such as Raft, Multi-Paxos, Zab and Viewstamped Replication that decide a sequence of values (a log).

They elect a leader per epoch and require a majority quorum both to elect it and to accept each proposal; any two majorities overlap, so a decision can never be lost or contradicted.

A number that increases every time the lock is granted, sent with every write to the protected resource.

The storage service rejects any write carrying a token older than one it has already seen, so a zombie client woken from a pause can't corrupt data. The resource itself has to check.

Small numbers in a Thrift or Protocol Buffers schema that identify each field in the encoded bytes instead of its name.

Tags make records compact and make renaming a field free. In exchange the tag is forever: you may never change a field's tag or reuse a removed field's tag, because old data still carries it.

Create many more partitions than nodes up front (e.g. 1,000 for 10 nodes); a new node steals whole partitions from existing ones.

The key-to-partition mapping never changes, only partition-to-node does, so moves are minimal. But the count is fixed at setup: too few and partitions grow huge, too many and overhead adds up.

In a fully asynchronous model with no clocks, no deterministic algorithm can always reach consensus if a node may crash.

Real systems escape it with timeouts (to suspect crashed nodes) or randomness. That is why election timeouts are randomized and why consensus can stall when the network is very unreliable.

A follower that restarts or falls behind asks the leader for every change after the last log position it processed.

Because the log has positions, a follower that was down can recover without a full copy. New followers start from a consistent snapshot and then catch up the same way.

Older code can read data that was written by newer code.

Harder, because old code must tolerate additions it could not have anticipated: it has to ignore unknown fields, and you can only remove fields that are optional or have a default.

Operation B happens after A if B knew about A, depended on it, or built on it. If neither knew about the other, the two are concurrent, whatever their wall-clock times.

When one operation happened before the other, the later one may overwrite the earlier. When they are concurrent, there is a real conflict and both values have to be kept or merged.

Disks crash, RAM goes bad, power and network fail. Mostly random and independent.

A disk's MTTF is 10 to 50 years, so a cluster of 10,000 disks loses about one a day. Redundancy (RAID, replicas, multi-machine setups) handles them, and software fault tolerance lets you lose whole machines.

An in-memory hash map from each key to the byte offset of its latest value in a data file.

Reads cost one seek and writes one append, which is ideal when there are many updates to few keys. The catch: all keys must fit in RAM, and range queries are impossible.

Pick the node as hash(key) mod number-of-nodes.

Don't: when N changes, most keys map to a different node (going from 4 to 5 nodes moves about 80% of keys), making rebalancing very expensive.

A few slow requests hold the server's limited workers, and the fast requests queued behind them wait.

This is why client-side response time can be far worse than server-side processing time. Load tests must send requests independently of responses (open loop), otherwise they hide the queue.

Relational-style operators (join, filter, group by) and SQL on top of the engines, instead of hand-written map and reduce functions.

Less code, and a query optimizer can choose the join algorithm (broadcast vs sort-merge) and push filters down. Unlike an MPP database, it still lets you run arbitrary code over arbitrary file formats.

A partition with disproportionately high load.

Hot spots come from the access pattern, not only the data size: sequential keys like timestamps send all of today's writes to one partition, and a celebrity's key is hot no matter how you hash it.

Operator mistakes, such as bad configuration changes, are a leading cause of outages.

The fixes are about design: interfaces that make the right thing easy, sandboxes, fast rollback, gradual rollouts to a small canary group, and detailed telemetry.

An operation that has the same effect whether it is performed once or several times.

Retries are the only way to cope with lost responses, and they are only safe if the operation is idempotent, e.g. by attaching a unique request ID the server deduplicates on.

An idempotent operation has the same effect whether it is applied once or several times; with it, at-least-once processing yields effectively exactly-once results.

Storing the input offset alongside the output, or committing output and offset atomically, makes reprocessing after a crash harmless.

Batch jobs write fresh files (search indexes, key-value database files) rather than mutating a live database; readers switch to the new output atomically.

Bad code costs a re-run or a rollback, not corrupted data (human fault tolerance). Building files in bulk is also far faster than one network write per record, and it doesn't overload production databases.

Code that spells out every step and its order, such as a loop over records.

Hard to parallelize and hard for the system to optimize, because the code fixes the algorithm.

A participant that has voted yes but not yet heard the decision: it can neither commit nor abort.

It must hold its locks until the coordinator recovers, blocking other transactions. Orphaned in-doubt transactions sometimes need a human (heuristic decision) to resolve, which risks breaking atomicity.

An additional structure derived from the primary data that helps locate records quickly.

Indexes speed up reads but slow down every write, since each one must also update the index. That trade-off is why databases don't index everything by default.

Human-readable textual encodings that are widely supported across languages.

Good enough as a lingua franca between organizations, but verbose, ambiguous about numbers (integers above 2⁵³ break in JavaScript), lacking binary strings, and schema support is optional and complicated.

A batch layer periodically recomputes exact views from immutable events; a speed layer serves fast approximate updates in between.

It combined reliable batch reprocessing with low-latency streams, but you implement the logic twice in two frameworks and must merge their outputs, and the approximate path drifts between batch runs.

A (counter, node id) pair; every node sends its counter on every message and, on receipt, sets its counter to max(local, received) + 1.

They give a compact total order that is consistent with causality: if A happened before B, A's timestamp is smaller. The converse fails: a smaller timestamp does not mean the event came first causally.

Built-in serializers such as Java Serializable, Python pickle or Ruby Marshal.

Convenient but they tie your data to one language, often allow decoding to instantiate arbitrary classes (a security hole), and treat versioning and efficiency as afterthoughts. Avoid them for anything that persists or crosses a team boundary.

Each write carries a timestamp, the highest one is kept, and the others are silently discarded.

Replicas converge, but at the cost of durability: concurrent writes are lost, and with clock skew even a write that really happened later can lose.

Resolving conflicts by keeping the write with the greatest wall-clock timestamp.

A node whose clock is a few ms slow writes timestamps smaller than writes it has already seen, so its causally later writes are silently dropped. Logical clocks order events by causality instead.

The client (or a coordinator) sends every write and every read to several replicas in parallel. No node is in charge of ordering writes.

There is no failover, and a few slow or dead nodes are tolerated. Stale replicas are repaired by readers and by background processes.

A lock with a timeout: the holder must renew it before it expires, or someone else may take over.

Leases keep a crashed holder from blocking everyone forever, but a paused holder doesn't know its lease has run out and can act on stale authority.

Make a replicated object behave as if there were one copy and every operation took effect atomically at a single point between its start and end.

It is a recency guarantee: once any read returns a new value, all later reads (in real time) must too. Checking it means finding a total order of operations that respects real time and register semantics.

Load balancing delivers each message to one of several consumers; fan-out delivers every message to all of them.

Load balancing parallelizes expensive processing; fan-out lets independent consumers each see the whole stream. Consumer groups in log-based brokers combine both.

The few numbers that describe load for your system: requests/s, read/write ratio, active users, cache hit rate.

Choosing the right one matters. For Twitter the tweet rate was not the bottleneck; fan-out (followers per user) was.

Periodically discarding older log entries for a key, keeping only the most recent value (and dropping keys with a tombstone).

The log then stays proportional to the size of the database rather than to its write history, yet a new consumer can still rebuild the full current state by reading it from offset 0.

A broker that stores messages in an append-only, partitioned log on disk; consumers read sequentially and messages are not deleted on consumption.

It gives durable, replayable, totally ordered (per partition) streams with high throughput. Parallelism is at the granularity of partitions, and a slow message holds up its whole partition.

Two transactions read-modify-write the same value concurrently, and one overwrites the other's change.

The classic counter-increment bug. Fix it with atomic updates, explicit locks (SELECT FOR UPDATE), compare-and-set, or a database that detects it and aborts.

A storage engine built from a memtable plus a cascade of SSTables that are merged and compacted in the background.

It turns random writes into sequential ones, giving high write throughput and good compression. In exchange, reads may check several tables and compaction competes with live traffic.

Links in both directions, such as people and the organizations they worked for.

As applications grow, data tends to become more interconnected. That pushes document models toward references and joins, or toward graphs.

Many records pointing at one shared entity, such as many people living in one region.

They need joins. Relational databases do joins well; with documents the application often has to emulate them with extra queries.

A programming model and framework: a user-defined mapper extracts key-value pairs from each record, the framework sorts and groups them by key, and a reducer processes all the values for each key.

You write two callbacks that each see a small piece of the data, and the framework handles parallelism, moving data between machines and retrying failures. Each job is a single map-sort-reduce, so real work is chained into workflows.

A query written as a pure map function that emits key/value pairs and a reduce function that combines the values for each key.

Somewhere between declarative and imperative. Pure functions can run anywhere in any order, but writing two coordinated functions is harder than one query, which is why MongoDB added the declarative aggregation pipeline.

Writing each job's full output to the distributed filesystem, replicated, before the next job may start.

It makes fault tolerance simple (just rerun the failed task), but every intermediate dataset is written and replicated, and a stage cannot start until the slowest task of the previous one is done.

A stored copy of a query's results (e.g. a grid of aggregates by dimension) that's updated when the underlying data changes.

Common aggregates become lookups instead of scans, but every write gets more expensive and the cube can't answer queries on dimensions it doesn't keep.

An in-memory balanced tree (e.g. red-black or skip list) that buffers recent writes in sorted order.

When it exceeds a threshold it is written out as a new SSTable in one sequential pass. A write-ahead log protects it against crashes.

An intermediary that stores messages from producers and delivers them to consumers asynchronously.

It buffers when the recipient is slow or down, redelivers after crashes, decouples sender from recipient (sender doesn't need its address), and fans one message out to many consumers. The sender doesn't wait for a reply.

A server that producers send events to and consumers receive them from, decoupling the two.

Brokers tolerate clients coming and going and absorb bursts. Two key design questions: what happens when producers are faster than consumers, and what happens if a node crashes.

Breaking a stream into small blocks processed like mini batch jobs, or periodically checkpointing operator state and input offsets.

On failure the processor restarts from the last checkpoint and discards partial output, but any side effects already sent to external systems would be repeated.

A clock guaranteed to move only forward, used to measure durations such as timeouts.

Its absolute value means nothing and can't be compared across machines, but the difference between two readings on one machine is a reliable elapsed time.

If a user has read a value, later reads never return an older one.

Reads that hit a fresh replica and then a stale one make time appear to move backwards. Pinning each user to one replica (for example by a hash of the user ID) prevents this.

Several nodes accept writes (for example one leader per datacenter) and exchange their changes asynchronously.

Writes are local in each datacenter and the system survives a datacenter or network outage. In exchange, the same record can be changed concurrently in two places, and those conflicts have to be resolved.

Keep several committed versions of each row, tagged with created_by and deleted_by transaction IDs, so different transactions can see different points in time.

Visibility rules decide which version each reader sees, which is how snapshot isolation is implemented without read locks; garbage collection removes versions nobody can see.

Part of the network is cut off from the rest, while each side keeps working.

Only the side holding a majority can keep electing leaders and committing writes. Partitions are surprisingly common, even inside one datacenter, so software must handle them deliberately.

Storing each human-meaningful fact once and referring to it by ID everywhere else.

Avoids duplicated text that can drift out of sync and makes renames a single write. The price is a join (or an extra lookup) when reading.

The awkward translation layer between objects in application code and tables of rows.

ORMs reduce the boilerplate but don't hide the difference completely. A JSON document is closer to the objects many applications use.

Transaction processing (few records by key, low-latency writes) vs analytics (scans and aggregates over huge numbers of records).

The two access patterns differ so much that they run on separate systems, optimized differently for each.

One record owning many children, like a user's positions, schools and contact links.

These form a tree. Documents store the tree directly; relational schemas spread it over several tables joined by a foreign key.

Making it easy for operations teams to keep the system running smoothly.

Good monitoring, automation, documentation, predictable behaviour and sensible defaults let people focus on the high-value work.

A unique request id minted by the client and carried all the way to the database, enforced with a uniqueness constraint.

Retrying is the only way to survive a lost response, and only an end-to-end id makes a non-idempotent operation like a money transfer safe to retry: the second attempt finds the id and does nothing.

Every change is appended to one totally ordered log; each derived system consumes it in that order.

Ad-hoc dual writes from the application race and partially fail, leaving stores permanently inconsistent. With a log, all consumers apply the same writes in the same order, so they can only be behind, never contradictory — they converge.

When an insert overflows a full page, the page is split in two halves and the parent gets a new key and reference.

Splits keep the tree balanced and grow it upward one level at a time, but they write several pages at once and leave half-empty pages (fragmentation).

Analytic (MPP) databases break complex queries into stages that run on many partitions at once.

Key-value style single-partition reads are the simple case; warehouses push joins, filters and aggregations down to partitions in parallel (the topic returns in chapter 10).

Some parts of the system are broken in unpredictable ways while others keep working.

A single computer is designed to either work or crash cleanly; a distributed system is nondeterministic, and you may not even know whether an operation succeeded. Fault tolerance has to be designed in from the start.

A subset of the data such that every record, row or document belongs to exactly one partition.

Partitioning exists for scalability: different partitions live on different nodes, so data volume and query load are spread over many disks and CPUs. It is usually combined with replication, so each partition has copies on several nodes.

A map-side join for two inputs already partitioned the same way on the join key: mapper i joins partition i of each input.

Each mapper only needs one partition of the smaller side in memory. It relies on how the inputs were written earlier, so it needs the partitioning recorded as metadata (Hive bucketed map joins).

A hash function maps each key to a number; each partition owns a range of hashes.

A good hash spreads even skewed keys uniformly, avoiding hot spots for many patterns. The cost is losing the sort order: a range query has to be sent to all partitions.

Each partition owns a contiguous range of keys, like volumes of an encyclopedia; keys are kept sorted within a partition.

Range scans are cheap because neighbouring keys live together. But boundaries must adapt to the data, and access patterns that cluster on adjacent keys (timestamps) create hot spots.

A fixed number of partitions per node; a joining node splits a random selection of existing partitions and takes half of each.

Partition size stays stable as the cluster grows. Random split points can be unfair, but with enough partitions per node (Cassandra used 256) the load averages out.

The response time that a given share of requests is faster than. The median (p50) is the typical case.

The mean doesn't tell you how many users waited. High percentiles (tail latencies) hit the users with the most data, often the most valuable customers.

A write in one transaction changes the result of a search query in another, typically by inserting a row that didn't exist when the search ran.

There's no row to lock when the check is 'no booking exists'; you need predicate or index-range locks, materialized conflicts, or SSI.

Configure an initial set of partition boundaries on an empty database.

It avoids the single-partition bottleneck at the start, but only works if you can predict the key distribution in advance.

Locks on all objects matching a search condition, including rows that don't exist yet; usually approximated by locking a range of an index.

What lets 2PL prevent phantoms such as double-booked meeting rooms.

Automated decisions (loans, insurance, hiring, policing) driven by models trained on data.

Models can codify existing discrimination and create feedback loops, and the people affected rarely can appeal or even learn why. Engineers who build these systems share responsibility for their consequences.

A graph processing model: in each superstep every vertex processes messages sent to it, updates its state, and sends messages along its edges.

Iterative graph algorithms like PageRank don't fit one MapReduce pass. Pregel keeps vertex state in memory between iterations, and a checkpoint after each superstep gives fault tolerance.

When old code reads a record, modifies it and writes it back, keeping the fields it didn't understand.

If old code decodes into a model object that drops unknown fields and then re-encodes, it silently destroys data written by newer code — without any error. Protobuf's generated classes keep unknown fields; many ORMs do not.

Behavioural data collected for one feature can reveal far more, and lives on beyond its purpose.

Every stored record can be breached, subpoenaed or repurposed after an acquisition. Collect less, keep it shorter, and treat data as a hazardous material rather than an asset.

A thread stopped for an arbitrary time without noticing: stop-the-world GC, VM suspension, context switches, disk I/O, swapping, SIGSTOP.

Code that checks a condition and then acts can be frozen between the two steps. When it wakes up, the world, and its lease, may have moved on.

Vertices and edges, each with a label and key-value properties; any vertex can connect to any other.

Very flexible for highly connected, evolving data. New kinds of vertices and edges need no schema change, and traversals follow edges in either direction.

With n replicas, a write must be confirmed by w of them and a read must query r of them. If w + r > n, every read set overlaps every write set.

Overlap means at least one queried replica has the latest successful write. There are still edge cases: sloppy quorums, concurrent writes, failed writes that are not rolled back, and a node restored from a stale replica.

Isolation level that prevents dirty reads (show the last committed value) and dirty writes (row write locks).

The default in PostgreSQL, Oracle and SQL Server. It still allows read skew, lost updates and write skew.

Read repair: a client that sees a stale response writes the newer value back to that replica. Anti-entropy: a background process compares replicas and copies the missing data.

Read repair only fixes values that get read. Without anti-entropy, rarely read keys can stay stale on some replicas indefinitely.

A transaction sees different parts of the database at different points in time, e.g. one account before a transfer and the other after.

Tolerable for a single page view, fatal for backups and long analytic queries, which would observe a state that never existed.

A user always sees their own updates after a page reload. Other users' updates may still be delayed.

Without it, a user who has just posted something reloads the page and the post seems to be gone. Fixes include reading recently modified data from the leader, or tracking the user's last write position and serving reads only from replicas that have caught up to it.

Logging read requests into a stream and answering them with a stream–table join against the derived state.

It records exactly what each user saw and when, which helps tracking causality and auditing; it also lets reads be routed to the partition holding the data, like a distributed query.

Moving load from one node to another as data, query rate or the set of machines changes.

Afterwards load should be fair, the database should keep serving reads and writes during the move, and no more data should move than necessary.

Fault tolerance without materialization: lost data is rebuilt by rerunning the operators that produced it, starting from input that is still available.

Avoids writing intermediate state at all, but a failure late in a long pipeline can mean a lot of recomputation, and operators must be deterministic so the rebuilt data matches what was lost. Checkpoints bound the cost.

Mappers emit the join key from both inputs, the shuffle brings all records with that key to one reducer, and the reducer joins them.

It needs no assumptions about the inputs, and each reducer only holds one key's records at a time. The price is sorting, copying and merging every input record.

Data organized as relations (tables) of unordered tuples (rows), queried with SQL.

It hides how the data is stored behind a clean interface. The query optimizer, not the developer, decides the access path, so new queries don't need new code paths.

The system keeps working correctly, at the expected performance, even when things go wrong.

Things going wrong is the normal case at scale. Reliability is designed in, by expecting faults and keeping them from turning into user-visible failures.

Making a request to a remote service look like calling a local function.

The abstraction is leaky: a network call can be lost, time out with an unknown outcome, take wildly variable time, and be retried — so without idempotence a retry may run the action twice. Treat a remote call as a different kind of thing, not as a function.

The delay between a write on the leader and that write appearing on a follower. In normal operation it is under a second, but under load or during failures it can be minutes.

Asynchronous followers converge only eventually, and the delay has no upper bound. Applications that read from followers have to decide which anomalies they can tolerate.

The stream of changes a leader sends to its followers. It can be statement-based, write-ahead-log shipping, a logical (row-based) log, or trigger-based.

Replicating statements breaks on nondeterminism (NOW(), RAND()). Shipping the WAL ties followers to the leader's storage format. A logical row log is decoupled from the storage engine and can feed change data capture.

Building a new version of a derived view by replaying the full history alongside the old one.

It turns schema migrations into a gradual, reversible process: run both views, shift a fraction of reads to the new one, and roll back at any point because the old view never stopped.

How a client finds the node that holds the key it wants: ask any node, go through a partition-aware routing tier, or use a partition-aware client.

It is service discovery for partitions. The hard part is everyone agreeing on the assignment while partitions move; a stale view sends requests to the wrong node.

Response time is what the client sees (network + queueing + service). Latency is the time a request spends waiting to be handled.

Measure response time at the client. Time measured only inside the server misses the queueing delay.

A design style built on HTTP's own features (URLs, methods, caching, content types), usually with JSON bodies.

Easy to experiment with and debug, supported everywhere, and good for public APIs. Compatibility is maintained by versioning the API, often in the URL or headers, because you can't force clients to upgrade.

Deploying a new version to a few nodes at a time, so old and new code run side by side for a while.

It enables zero-downtime deploys and quick rollbacks, but it means data flows in both directions between versions, so both backward and forward compatibility are required.

Safety: nothing bad ever happens. Liveness: something good eventually happens.

A safety violation happens at a specific moment and can't be undone, so it must hold in every situation. Liveness may come with caveats, such as only once the network has recovered.

Whether there are reasonable ways to deal with growth in load, data volume or complexity.

It is not a yes/no label. The useful question is: if load grows in this particular way, what are our options for adding resources?

Vertical scaling moves to a more powerful machine; horizontal scaling spreads load over many small machines (shared-nothing).

Big machines get expensive fast and are a single point of failure; distributing stateless services is easy, but distributing stateful data adds a lot of complexity. Real systems mix the two.

Send a query to all partitions in parallel and combine their answers.

The query is only as fast as the slowest partition, so tail-latency amplification makes scatter/gather reads expensive, even when they run in parallel.

Changing a schema over time (adding, removing, renaming, retyping fields) while keeping old and new data mutually readable.

Each format has its own rules: tag-based formats tolerate renames but not tag changes; Avro tolerates reordering and tag-free generation but needs aliases for renames and defaults for any field it adds or drops.

Changing a schema, e.g. ALTER TABLE followed by an UPDATE that rewrites existing rows.

Adding a column is usually fast, but rewriting every row of a large table can take a long time and, in some databases, locks the table.

Schema-on-read: the structure is implicit and interpreted when data is read. Schema-on-write: the database enforces an explicit schema.

It's like dynamic vs static typing. Schema-on-read lets old and new formats coexist and handles heterogeneous data; schema-on-write needs migrations but guarantees the shape.

Arranging reducer input so that, within one key, records arrive in a chosen order, e.g. the profile record first and then events by timestamp.

With the record you join against arriving first, the reducer can stream the rest past it with constant memory instead of buffering them.

The log is split into fixed-size segment files; frozen segments are compacted (keeping only the latest value per key) and merged in the background.

This reclaims space taken by overwritten values without ever modifying a file in place. Reads and writes continue on the old segments until the merged one replaces them.

One follower is synchronous and the rest are asynchronous. If the synchronous one becomes slow, an asynchronous follower takes its place.

This guarantees an up-to-date copy on at least two nodes without letting a single straggler stop all writes.

The strongest isolation level: the result is the same as if transactions had run one at a time, in some order.

Prevents every race condition above. Three ways to get it: actual serial execution, two-phase locking, and serializable snapshot isolation.

Snapshot isolation plus tracking of which transactions read data that a concurrent transaction wrote; on a dangerous pattern, abort at commit.

Optimistic: no blocking, readers and writers proceed freely. Performs well under low contention, but abort rates rise as contention grows.

Partitioning mapper output by a hash of the key, sorting it, and copying each partition from every mapper to its reducer.

It guarantees every value for a key reaches the same reducer, and it is the most expensive part of most jobs: an all-to-all copy over the network. Combiners shrink it by pre-aggregating on the mapper.

Accidental complexity comes from the implementation, not from the problem the users have.

Abstraction is the main tool for removing it: a clean interface hides implementation details, so a change behind it doesn't ripple through the codebase.

One replica (the leader) accepts all writes and streams them, in order, to read-only followers.

There is a single order of writes, so there are no write conflicts. The cost is that every write goes through one node, and reads from followers may be out of date.

Two policies for when and which SSTables to merge: by similar size, or into levels of non-overlapping key ranges.

Leveled compaction bounds how many tables a read touches and how much space garbage takes, at the cost of more rewriting. Size-tiered writes less but reads more.

Some partitions get more data or queries than others.

With skew, adding nodes stops helping: with 10 nodes but all load on one, you have the throughput of one node. The goal of partitioning is an even spread of both data and load.

A few keys (a celebrity's user id) have far more records than the rest, so one reducer gets much more work.

The job finishes only when the slowest reducer does. Skewed joins sample for hot keys, spread their records across several reducers, and copy the other side of the join to each of them.

When a key's home nodes are unreachable, writes are accepted by other nodes. Those nodes hand the data back to the home nodes once they recover.

Writes stay available during network problems, but w + r > n no longer guarantees a fresh read, because the w writes may land outside the n home nodes.

Service-level objectives and agreements, often stated in percentiles: e.g. p50 under 200 ms and p99 under 1 s.

Percentiles make the contract measurable: customers can ask for a refund if it is not met.

Each transaction reads from a consistent snapshot of the database as of when it started.

Readers never block writers and writers never block readers. Called 'repeatable read' in PostgreSQL and MySQL and 'serializable' in Oracle, though it isn't serializable.

Bugs and runaway processes that are correlated across nodes because every node runs the same code.

They cause many more simultaneous failures than hardware does (for example the 2012 leap-second bug that hung Linux servers). Replication doesn't help; testing, isolation, monitoring and crash-restart design do.

Counting by sorting so equal keys are adjacent, as opposed to holding a hash table of counts in memory.

A hash table is fast while the distinct keys fit in RAM. Sorting can spill to disk and merge sorted runs, so it scales past memory, which is why MapReduce is built around sorting.

The query language for RDF triple-stores, with property paths like :within*.

Even shorter than Cypher for the same query. Cypher's pattern syntax borrows from it.

An in-memory index holding only the first key of each block of a sorted file.

Because the file is sorted, you can jump to the right block and scan a few kilobytes. A few keys per block suffice instead of one entry per key.

Two nodes both believe they are the leader and both accept writes.

Their data diverges and there is no automatic way to reconcile it. Systems use fencing (STONITH, epoch numbers) to shut down or reject the deposed leader.

Append a random number (say two digits) to a very hot key so its writes spread over many sub-keys and partitions.

It relieves the hot partition, but every read must now fetch all sub-keys and combine them, and you need bookkeeping to know which keys are split. Most databases leave this to the application.

An immutable segment file whose key-value pairs are sorted by key, with each key appearing once.

Sorting lets segments be merged like mergesort, needs only a sparse index, and lets blocks be compressed, which removes most of a hash index's limits.

A central fact table of events, each row referencing dimension tables (who, what, where, when, how) by foreign key.

It's the standard warehouse model. Fact tables are huge and often very wide, while dimensions are small. A snowflake schema further normalizes the dimensions.

A document is stored as one contiguous value, so loading the whole thing is a single read.

It only pays off when you need most of the document. Reading one field still loads all of it, and updates usually rewrite all of it, so keep documents small.

A straggler is an event that arrives after its window was believed complete; a watermark is the processor's declaration that no earlier events are expected.

You can never be sure all events for a window have arrived. Either ignore stragglers (and monitor how many you drop) or publish a correction later.

Consumers that perform side effects (emails, notifications, calls to external services) in response to events.

Unlike pure views, side effects don't replay safely: rebuilding from offset 0 sends every email again. Side-effecting consumers need to start at the head or deduplicate on event ids.

Joining events from two streams that share a key and occur within some window of each other, e.g. searches and clicks.

The processor must keep state, an index of recent events from both sides, and the window decides when it stops waiting and emits a non-match.

Augmenting each event with data from a table, using a local copy of the table kept current by a changelog stream.

Avoids a remote database query per event. The join is time-dependent: an event joined before or after a profile update gets different data.

The leader confirms a write to the client only after a follower has also confirmed it.

An acknowledged write survives the loss of the leader. But one slow or failed follower blocks every write, so a fully synchronous setup is fragile.

An abstraction of the faults an algorithm must tolerate: timing (synchronous, partially synchronous, asynchronous) × node failures (crash-stop, crash-recovery, Byzantine).

An algorithm is only correct relative to a model. Partially synchronous with crash-recovery is the most realistic for most systems.

The authoritative copy of the data, where each fact is written first and exactly once.

When two copies disagree, the system of record wins by definition. Being explicit about which dataset is authoritative and which are derived is the first step to integrating many tools sanely.

Two changelog streams joined to maintain a materialized view of the join, such as a per-user timeline cache of tweets from followed accounts.

Instead of computing the join at read time, every tweet, delete, follow and unfollow incrementally updates the cached result.

When one user request makes N backend calls in parallel, it is as slow as the slowest call.

P(slow) = 1 − (1 − p)^N, so a 1% slow backend makes most user requests slow once N reaches the hundreds.

A secondary index that covers all data but is itself partitioned by the indexed term, separately from the documents.

A read needs only the partition holding that term. A single document write may now touch several partitions, so in practice the index is updated asynchronously and may briefly miss recent writes.

Some guarantees can only be implemented correctly with the help of the application at both endpoints of the communication.

TCP removes duplicate packets and a transaction is atomic, but neither knows that a retried HTTP request is the same transfer. Lower layers' guarantees help performance but don't remove the need for an end-to-end check.

Decisions such as declaring a node dead or choosing a leader are made by a quorum vote, not by one node.

A node can't trust its own view: it may be the one cut off, or the one that just paused. If a majority declares it dead, it must step down, even if it feels fine.

Schema-driven binary formats with code generation, from Facebook and Google respectively.

Thrift's CompactProtocol and Protobuf pack tag and type into one byte and write integers as variable-length varints, getting the example record down to 34 and 33 bytes.

Records processed or requests served per second.

Batch systems care mostly about throughput; online systems care mostly about response time. The two are linked through queueing.

A wall-clock reading (e.g. seconds since the epoch) kept in sync with NTP.

When NTP finds it too far off it can jump the clock backwards, so subtracting two readings can give a negative duration. It's fine for showing a date but wrong for measuring elapsed time.

Timeliness: readers see an up-to-date state. Integrity: no corruption, no contradictory or lost data.

Violations of timeliness are temporary — wait and they heal. Violations of integrity are permanent until someone repairs them. Most applications need integrity strictly and can tolerate some lag.

Declaring a node dead when it hasn't responded within a fixed period.

A short timeout detects crashes quickly but falsely kills slow nodes, whose load moves to others and can cascade. A long timeout leaves users waiting. There is no correct value, only a trade-off.

A special deletion record appended to the log in place of removing data.

Since nothing is edited in place, a delete has to be a write. Compaction later discards the tombstone and every older value it shadows.

A protocol that delivers every message reliably to every node, and in the same order on all of them.

It is state machine replication: apply the same deterministic operations in the same order and replicas stay identical. It is equivalent to repeated consensus, and lets you build linearizable compare-and-set on top of an append-only log.

Everything stored as (subject, predicate, object) statements, where the object is either a value or another vertex.

The same graph idea in a uniform format that merges easily. It's the basis of RDF and the semantic web.

Continuously checking the integrity of data — checksums, re-deriving views from the log, comparing replicas.

Disks rot, software has bugs, and checksums can vouch for a bug's output. Systems that are designed to be audited (immutable events, deterministic derivation) let you detect corruption early instead of trusting blindly.

A coordinator asks every participant to prepare (vote), then writes a commit or abort decision to its log and broadcasts it.

Two points of no return make it work: a participant that votes yes can no longer abort on its own, and the coordinator's logged decision is final. Its weakness is that it blocks if the coordinator dies.

Readers take shared locks and writers exclusive locks, all held until the transaction ends (grow, then shrink).

Pessimistic: readers block writers and writers block readers. Serializable, but with unstable latency, low throughput under contention, and frequent deadlocks.

Variable packet delay caused mostly by queueing: at switches, in the OS, at a busy CPU, under TCP flow control and retransmission.

Datacenter networks and the internet are packet-switched to use bandwidth efficiently, which means queues instead of reserved capacity. When load rises, the tail of the delay distribution grows without a hard limit.

Treating the log as the database's core and its indexes, caches and views as separate derived consumers.

A database already does this internally: CREATE INDEX scans existing data then follows new writes. Unbundling exposes that mechanism so you can compose specialized storage and processing tools, loosely coupled through the log.

Every program takes stdin and writes stdout as a stream of newline-separated records; MapReduce uses files in a distributed filesystem the same way.

A shared interface is what makes composition possible. It also separates logic from wiring: a program does not care where its input came from or where its output goes.

One stream processor handles both live events and the replay of history from the beginning of the log.

With a log that retains history and engines that support event time and exactly-once semantics, the same code can rebuild a view from scratch or keep it up to date — one implementation instead of two.

Route all requests for the same value to one partition and decide them sequentially in log order.

A single-threaded processor per partition decides deterministically which request came first, with no locks or distributed transactions, and scales by adding partitions.

Small programs that each do one thing well, composed with pipes into larger jobs.

Because every tool reads and writes the same thing, tools written decades apart compose freely, and you can inspect any intermediate stage by cutting the pipe. MapReduce is the same idea spread over many machines.

Integers encoded 7 bits per byte with a continuation bit, often after zigzag mapping so small negatives stay small.

Most numbers are small, so most take one or two bytes instead of a fixed eight. A side effect: widening a field from 32 to 64 bits is wire-compatible, but old 32-bit readers silently truncate large values.

One counter per node, merged element-wise on each message, so each event carries a summary of everything it has seen.

Unlike Lamport timestamps, vectors can tell whether two events are concurrent or causally ordered, at the cost of size growing with the number of nodes.

One version counter per replica, sent with every read and write. The server overwrites the values a write has seen and keeps concurrent values as siblings.

This tells overwrites apart from concurrent writes without depending on clocks, so no data is silently dropped. The application merges siblings, for example by taking the union of shopping carts (deletions need tombstones).

A total order of timestamps can pick a winner among conflicting claims only after all claims have been collected.

To enforce 'one owner per username' at the moment of the request, a node must know no other node is concurrently claiming it with a lower timestamp, which requires hearing from every node. You also need to know when the order is final: total order broadcast.

Tumbling (fixed, non-overlapping), hopping (fixed, overlapping), sliding (all events within an interval of each other), session (bounded by inactivity gaps).

Aggregations over an unbounded stream need a bounded scope. The window type determines which events are counted together and whether one event can count in several windows.

A chain of jobs where each job's output directory is the next one's input, managed by a scheduler that runs them in dependency order.

Most real analyses take dozens of MapReduce jobs. Since each output is a file, any stage can be rerun or inspected on its own.

The number of bytes written to disk per byte the application writes, because of WAL, page rewrites and repeated compaction.

In write-heavy workloads it's what caps throughput. LSM-trees usually amplify less than B-trees and, more importantly, write sequentially.

Two transactions read the same objects, each decides based on what it read, then they update different objects, together breaking an invariant.

A generalization of lost update that snapshot isolation does not prevent (doctors on call, double-spending, unique usernames). Needs serializable isolation or explicit locks on the rows read.

An append-only file where every modification is recorded before it is applied to the tree's pages.

A crash in the middle of a multi-page split would otherwise leave a corrupted tree. On restart the WAL restores consistency. LSM memtables use one for the same reason.

In Avro, the schema the data was encoded with and the (possibly different) schema the application expects when decoding.

Avro resolves the two side by side: fields are matched by name, so order doesn't matter; fields the reader doesn't know are skipped; fields the writer lacked get the reader's default. The writer's schema travels in the file header, a schema registry, or the connection handshake.

A standard C API for two-phase commit across heterogeneous systems such as databases and message brokers.

The coordinator is often a library inside the application, so if the application server dies, its log goes with it. XA must also be lowest common denominator: it cannot detect deadlocks across systems or support SSI.