Showing posts with label rdbms. Show all posts
Showing posts with label rdbms. Show all posts

Monday, 13 October 2025

Why Distributed Transactions Are Hard: The Hidden Cost of Scaling Relational Databases Horizontally

As applications scale and demand increases, developers often look to horizontally scale their databases to handle larger workloads. While this approach helps with performance and availability, it introduces a less visible but critical challenge: managing transactions across multiple nodes.

In a traditional, single-node relational database, maintaining ACID properties is straightforward, everything happens in one place. But when data and operations span across distributed systems, transactions become complex, fragile, and costly to orchestrate.

 

This post helps you to understand why distributed transactions are difficult, how they impact system performance under heavy load, and what tools or architectural patterns you might consider to overcome these limitations.

 

1. Horizontal Scaling

Horizontal scaling refers to adding more machines (nodes) to a system to distribute the load across multiple computing resources. Each node runs an instance of the application or database and participates in processing requests.

 

In the context of Databases,

 

Vertical Scaling: Increases capacity of a single database instance by upgrading hardware (CPU, RAM, IOPS), so it can serve more traffic/user requests. It is simpler to implement, and  minimal changes might be needed to the application code or DBMS configuration.

 

But Vertical Scaling is subject to physical hardware constraints. High availability may still require replicas, adding more hardware/procuring high end servers always incurs cost.

 

Horizontal Scaling: It involves distributing data and queries across multiple database instances. Following are the some common approaches used widely.

 

·      Read Replicas: multiple read-only replicas handle read queries, and the primary node handles writes.

·      Sharding: dataset is partitioned by shard key (e.g., user ID ranges), each shard resides on a different node.

·      Federated Databases: A federated database system integrates several independent, often geographically distributed, databases into one virtual database. This unified view allows users and applications to query data as if it resided in a single database, without needing to be aware of the underlying complexities or physical locations of the individual databases. In this model, each node is responsible for a subset of data, with coordination at the application layer.

 

1.1 Technical Challenges of Horizontal Scaling

Data Partitioning (Sharding): requires a shard key strategy that minimizes cross-shard queries. If the shard key is not desingned properly, then there is a risk of uneven data distribution (hot spots).

 

Distributed Transactions: ACID properties guarantees across multiple nodes require protocols like 2PC (Two-Phase Commit) or saga patterns, which add latency and complexity.

 

Data Consistency: Achieving strong consistency across replicas or shards is non-trivial. Here CAP Theorem applies, which states that consistency, availability, and partition tolerance cannot be fully achieved simultaneously in distributed systems.

 

Replication Lag: In asynchronous replication, data changes made to the primary database are not immediately reflected in the read replicas. As a result, applications reading from replicas might receive stale or outdated data. This can lead to inconsistencies, especially in systems that rely on real-time accuracy.

 

Joins are Expensive: When data is spread across multiple shards, performing operations like joins becomes challenging. Joins across shards are either very expensive in terms of performance or not supported natively by many databases. This limitation can slow down complex queries and affect the overall efficiency of the system.

 

To handle these challenges, applications often require extra logic at the application level or need to use distributed query engines. These tools help coordinate data across shards and replicas, but they also add complexity to the system architecture.

 

2. Distributed Transactions

A distributed transaction is a database transaction that spans multiple independent systems or databases (usually across different servers or nodes), and must be executed in a coordinated way so that either all operations succeed, or all fail, and ensures atomicity and consistency across systems.

 

2.1 ACID Transactions in a Single Node vs Distributed Environment

In a non-distributed relational database, all components reside within a single system. Write operations are typically handled by a primary node, while read operations can be offloaded to multiple replica (or "slave") nodes. The database engine centrally manages tables, indexes, logs, buffers, and locks.

 

When a transaction spans multiple tables, it is coordinated internally using local ACID properties there is no need for external coordination.

 

BEGIN;

  UPDATE accounts SET balance = balance - 500 WHERE id = 1;

  UPDATE accounts SET balance = balance + 500 WHERE id = 2;

COMMIT;

 

This entire transaction is atomic and isolated because it is executed within a single database engine on one machine.

 

Whereas in a distributed environment:

 

·      Atomicity: Every part of the transaction must either commit completely or not at all, spanning multiple nodes.

·      Consistency: All nodes must maintain a consistent state (Some systems support Eventual consistency), even in the event of failures.

·      Isolation: Concurrent transactions must be properly managed across distributed resources.

·      Durability: Once committed, a transaction’s changes must be permanently recorded across multiple systems.

 

2.2 Why it is difficult to implement Distributed Transactions?

 

Network Latency and Failures: Implementing distributed transactions is difficult largely because they depend heavily on network communication between multiple, physically separate nodes. Unlike a single-server system where data and transaction coordination happen internally with minimal delay, distributed systems must send messages back and forth over a network. This introduces network latency, meaning every step in the transaction like preparing, committing, or rolling back changes on different nodes takes longer simply because of the time it takes for data to travel across the network. Moreover, networks are inherently unreliable, packets can be lost, delayed, duplicated, or nodes can become temporarily unreachable due to hardware failures, maintenance, or congestion. These uncertainties raise the risk of partial failures where some nodes successfully commit their part of the transaction, while others do not. Handling such inconsistencies requires complex coordination protocols and recovery mechanisms, which add overhead and further increase latency. Together, network delays and the possibility of failures make maintaining atomicity and consistency across distributed transactions a challenging problem in distributed systems.

 

Two-Phase Commit (2PC): It is a classic protocol designed to ensure that distributed transactions maintain ACID properties (atomicity, consistency, isolation, and durability) across multiple database nodes. It works in two main stages:

 

·      First, in the prepare phase, the coordinator sends a request to all participating nodes asking if they are ready to commit the transaction. Each node performs all necessary checks and writes the transaction changes to a temporary log but does not finalize the commit yet. They then respond with either a “yes” (ready to commit) or “no” (abort). Only if all nodes reply “yes” does the coordinator proceed to the second stage,

 

·      Second the commit phase, where it instructs all nodes to finalize and permanently commit the changes. If any node fails to commit, the coordinator tells all nodes to roll back the transaction or some manual intervention might be needed.

 

While 2PC guarantees atomicity across nodes, it has some significant drawbacks. Because all nodes must wait for each other’s responses, the protocol can become slow under heavy load, as network communication and disk writes add latency. Moreover, 2PC is a blocking protocol, if any node or the coordinator crashes after the prepare phase but before sending the commit or rollback message, the other nodes remain locked in a waiting state, unable to proceed or release resources. This can cause the entire system to stall unless complex recovery mechanisms or timeouts are implemented, which adds further complexity to distributed transaction management.

 

Concurrency and Locking: When a database is distributed across multiple nodes, managing concurrency, the ability for multiple transactions to run simultaneously without interfering with each other becomes significantly more challenging. Each node must enforce locks on the data it manages to ensure transactions don’t overwrite or read inconsistent data. As the number of nodes increases, so does the total number of locks held across the system. This higher volume of locks raises the chances of deadlocks. Additionally, contention increases as multiple transactions compete for the same resources, causing some to wait longer before they can proceed. This waiting leads to blocking, where transactions are forced to pause until the necessary locks are released. Together, these issues reduce the system’s overall throughput (the number of transactions completed in a given time) and add complexity to concurrency control mechanisms. Distributed systems must implement efficient algorithms to detect and resolve deadlocks, manage lock timeouts, and optimize resource access, making concurrency control a major challenge as the system scales horizontally.

 

Consistency Models: Maintaining strong consistency across distributed nodes means ensuring that every node always sees the same up-to-date data at any given time. Achieving this requires constant synchronization between all the nodes so that updates are immediately reflected everywhere. However, in a distributed system, network delays, failures, or partitions, where nodes become temporarily unreachable make perfect synchronization difficult.

 

According to the CAP theorem, a distributed system can only guarantee two out of following three properties at the same time.

 

·      Consistency,

·      Availability, and

·      Partition tolerance.

 

Prioritizing strong consistency often means sacrificing availability because the system must wait for all nodes to synchronize before responding to requests. Alternatively, if the system remains available during network partitions, it may have to allow some nodes to operate on stale data, sacrificing strict consistency. This fundamental trade-off means that maintaining strong consistency in distributed systems is complex and often involves balancing synchronization overhead, response times, and fault tolerance.

 

In summary, we need an effective and robust distributed transaction manager (TM), which is a critical component for ensuring reliable coordination of transactions that span multiple nodes or services in a distributed system. The TM’s primary responsibility is to orchestrate the entire lifecycle of a distributed transaction, making sure it adheres to ACID properties across all participants. To do this, the TM communicates with all involved resource managers (such as different databases or microservices), coordinating their actions to prepare, commit, or roll back changes in a way that maintains global consistency.

 

The TM must also handle failure recovery gracefully. This involves detecting and resolving partial failures where some nodes may have committed while others failed, as well as managing situations where the TM or participants crash during the transaction process. It typically uses durable logs to record transaction states and decisions, enabling it to resume or roll back incomplete transactions after a failure.

 

In addition, the TM often implements standard protocols like Two-Phase Commit (2PC) or more advanced consensus algorithms to ensure reliable communication and coordination. It abstracts the complexity of distributed coordination away from individual resource managers, providing a unified interface that guarantees transactional integrity, even in the case of network delays, node failures, and concurrent access conflicts.

                                                                                System Design Questions

Understanding ACID Properties in Database Systems

In database systems, maintaining the correctness and reliability of data during transactions is essential. The ACID properties (Atomicity, Consistency, Isolation, and Durability) define a set of standards that ensure transactions are processed accurately and reliably. This blog post explains each of these properties with clear definitions and examples to help you understand how they contribute to data integrity and system reliability in transactional databases.

 

1. Atomicity

Atomicity refers to the "all-or-nothing" nature of a database transaction. In simple terms, this means that a transaction must be fully completed or not executed at all. If any part of the transaction fails, the entire operation is rolled back, and the database remains unchanged.

 

This property ensures that there is no partial execution of a transaction. It prevents scenarios where only some operations in a transaction are applied, leaving the database in an inconsistent or incomplete state.

 

1.1 Why is Atomicity important?

In real-world applications, a transaction often involves multiple steps or operations that are logically connected. These steps must be treated as a single unit of work. If one step fails, none of the steps should take effect. This guarantees that the system's state remains valid and predictable.

 

Consider a simple fund transfer between two bank accounts A and B:

 

·      Debit 5,000 from Account A.

·      Credit 5,000 to Account B.

 

These two operations must occur together. If the system debits Account A but fails to credit Account B (due to a crash, network issue, or any error), the customer would lose money, and the bank records would be incorrect. With atomicity, such a transaction would be rolled back entirely if both operations cannot be completed successfully.

 

1.2 How is Atomicity Achieved in Database Systems?

database systems use three core mechanisms to achieve Atomicity.

 

·      Transaction Logs

·      Rollback

·      Commit

 

Let's understand this with a successful money transfer from Account A to Account B.

 

Let’s say a customer initiates a transaction to transfer 5,000 from Account A to Account B using a banking app. This transfer involves two critical updates.

 

·      Debit 5,000 from Account A

·      Credit 5,000 to Account B

 

Step 1: Transaction Begins

The database system starts a new transaction and assigns it a unique transaction ID (e.g., TX1234). This marks the beginning of the atomic unit of work. Nothing is changed in the actual database yet.

 

Step 2: Validate the Operations

The database performs preliminary validations to avoid logging unnecessary or invalid transactions.

 

It checks:

·      If Account A has sufficient balance.

·      If the accounts involved are valid and active.

·      If the debit and credit operations are logically sound.

·      If the transaction respects all integrity constraints.

 

Step 3: Write-Ahead Logging (Transaction Log)

Before any change is written to the database, the system writes detailed information about the transaction to a transaction log.

 

This includes:

·      Transaction ID (TX1234)

·      Operation types (DEBIT, CREDIT)

·      Before and after values

·      Current status (PENDING)

 

This is known as Write-Ahead Logging (WAL), means the changes must be logged before they are applied. This allows the system to recover if a failure occurs. For high reliability, some databases replicate this log to another disk or standby server. This protects against disk crashes and ensures durability.

 

Step 4: Execute the Operations

After logging, the debit and credit operations are carried out in memory, no actual disk operation occurs yet.

 

Step 5: Commit

If all operations succeed:

 

·      The changes are written from memory to the actual database (on disk).

·      A COMMIT record is added to the transaction log. The transaction is marked as successfully completed.

 

At this point, the updates are permanent and visible to other users. The atomicity principle is honored, both debit and credit happened together as a single unit.

 

2. Consistent

Consistency means that a database must always transition from one valid state to another valid state after any transaction completes. The data should never end up in a state that violates the defined rules, constraints, or logical relationships of the database.

 

2.1 What Does Consistency Really Mean?

Imagine the database as a system that follows strict rules about how data relates to each other. These rules include:

 

·      Data integrity constraints like primary keys, foreign keys, and unique constraints.

·      Business rules such as "an order must always belong to an existing customer."

·      Validation rules ensuring data formats, ranges, or mandatory fields are correct.

 

A consistent database means every transaction preserves all these rules.

 

Example: Deleting a Customer and Their Orders

Suppose you have two tables:

 

·      Customers: stores customer information.

·      Orders: stores all orders placed by customers, linked by a foreign key to the Customers table.

 

If a transaction deletes a customer record, consistency requires that:

·      All their orders must also be deleted (cascading delete), or

·      The system must prevent the deletion if orders exist (restrict delete).

 

2.2 Why Is This Important?

If the system allowed deleting a customer but left behind orders that reference that now-nonexistent customer, the database would enter an inconsistent state:

 

Orders would have foreign keys pointing to a missing customer.

Applications reading these orders could crash or behave unpredictably.

 

2.3 How Is Consistency Enforced?

Database systems use several mechanisms to ensure consistency:

 

·      Constraints: Foreign keys, unique keys, check constraints.

·      Triggers: Automatically execute rules or actions to maintain consistency.

·      Transaction rules: The entire transaction must commit or rollback as a whole, so partial updates can’t leave data inconsistent.

 

In summary, Consistency means that after every transaction, the database stays correct and follows all the rules. It makes sure the database never shows wrong or confusing information to people or programs, so the data can be trusted and is reliable.

 

3. Isolation

Isolation means that when two or more transactions happen at the same time, they don’t get mixed up with each other. Each transaction works by itself, without interfering with others.

 

For example, if two transactions try to change the same piece of data at the same time, one of them will wait until the other finishes. This way, the changes don’t cause mistakes.

 

Isolation keep the data accurate, even when many users are using the database at once.

 

3.1 Why is Isolation Important?

When many people or programs use a database at the same time, they often try to read or change the same data simultaneously. Without isolation:

 

·      Changes made by one transaction could mix up or conflict with changes from another.

·      This can cause errors, like incorrect balances, lost updates, or corrupted data.

 

For example, if two people try to withdraw money from the same bank account at the same time, without isolation, both might see the old balance and withdraw too much.

 

Isolation keeps each transaction separate and protected, so they don’t interfere with each other. This helps the database stay accurate and reliable even when many users work at once.

 

3.2 How is Isolation achieved?

Databases use different techniques like Locking, Transaction Scheduling, Isolation Levels etc., to keep transactions isolated.

 

3.2.1 Locking

When a transaction wants to read or write data, the database locks that data so others can’t change it at the same time.

 

For example, if Transaction A locks a customer’s account to update the balance, Transaction B must wait until Transaction A finishes.

 

Locks can be:

·      Shared locks for reading (many can read at once).

·      Exclusive locks for writing (only one can write at a time).

 

3.2.2 Transaction Scheduling

The database controls the order in which transactions run, making sure they don’t overlap in ways that cause conflicts. It may delay some transactions so that each one appears to run alone, even if they actually run at the same time.

 

3.2.3 Isolation levels

Isolation levels decide how strictly transactions are kept separate from each other when they run at the same time. The choice of level affects both the accuracy of data and the speed of the system.

 

Higher isolation levels keep transactions very separate and avoid almost all conflicts, but they can slow things down because transactions might have to wait longer.

 

Lower isolation levels let transactions overlap more, which can speed things up, but may sometimes allow incorrect or confusing data to appear temporarily.

 

Following are the some common isolation levels:

·      Read Uncommitted

·      Read Committed

·      Repeatable Read

·      Serializable

 

Read Uncommitted

Transactions can see uncommitted changes made by other transactions ("dirty reads") No locking of read data. It offers the highest performance but lowest consistency

 

It is rarely used in practice except for approximate analytics where absolute accuracy isn't critical

 

Read Committed

Transactions can only see committed changes (no dirty reads). It internally use row level locks for writes. Each read operation sees only data committed before that specific read. Following are possible:

 

·      Non-repeatable reads (a row read twice in same transaction may differ if another transaction committed changes). For example, you read an account balance twice during a transaction and see different amounts because another transaction updated it in between.

 

·      Phantom reads (new rows may appear in subsequent reads)

 

Repeatable Read

The Repeatable Read isolation level ensures that if a transaction reads a row once, it will see the same data if it reads that row again later during the same transaction. It uses locks to prevent other transactions from modifying data that has been read.

 

It allows Phantom reads, like if a transaction runs a query like SELECT * FROM orders WHERE amount > 100, and another transaction inserts a new matching row before the first one commits, the new row can appear in repeated queries.

 

Serializable (Highest Isolation)

Transactions run so strictly that it looks like they run one after another, not at the same time. It prevents all types of conflicts, including dirty reads, non-repeatable reads, and phantom reads. It is most secure but slowest, because transactions often wait for others to finish.

 

It is used when absolute accuracy is required, like in banking systems.

 

 

4. Durability

Durability means that once a transaction is successfully completed (committed), its changes are permanent, they will not be lost, even if the system crashes right after.

·      The data is safely stored on disk.

·      It will still be there if the server restarts, power goes out, or the database crashes.

 

Following core mechanisms helps to ensure Durability.

 

Write-Ahead Logging (WAL):  Before applying any changes to the actual database, the database writes a log record describing the changes. In general, this log record is replicated accross multiple machines to ensure durability.

 

Checkpointing: A checkpoint is a snapshot operation that synchronizes in-memory data with disk storage.

 

In Distributed systems, they have Quorum-Based approach, where the durability is achieved via replication. Here a write is only considered successful if multiple nodes acknowledge it.

 

                                                                                System Design Questions