DST Lezione 11-12

  • Source: Fundamentals of Database Systems | Elmasri - Navathe | Chapter 23

A distributed computing system consists of a number of processing sites or nodes that are interconnected by a computer network and that cooperate in performing certain assigned tasks.

From Distributed computing systems, Distributed Database are born. They are defined as a “collection of multiple logically interrelated databases distributed over a computer network”.

Each different node is connected and the user just see as it was a single database

Replication

Replication means that copies of the same data is stored at multiple nodes for benefits like availability, reliability, geographical distribution and so on.

How much replication do i want from data? Let’s see two extreme scenarios

Scenario 1 - Full Replication - All data in all the nodes

I have k nodes, and each node have a copy of the data. I can tolerate up to complete failures of my node and i have the maximum possible reliability in the system.

In terms of memory, it’s more expensive because i use more memory. In terms of fastiness of retrieving data (performance), it’s generally better. Kind of like indexes, the problem is when you write on data; for example if i have to delete or update a record, i have to delete it from all the k nodes.

In general the problem with full replication is the overhead of the synchronization of data across nodes and various problem that arise because multiple client could write on multiple nodes on the same data.

Another problem concern the failure to updating a value. If i update a record, i need to update separate values on the physical layer but one of these could fail. Generally update are transaction (ACID properties), so when a transaction fail it should be reversed.

Managing full replication also requires a very sofisticated and a distributed transaction manager, that is obviously centralized. But using a centralized system for managing parallelism in a distributed system is crazy like why you’re even using a distributed system.

Scenarion 2 - Zero Replication

I have nodes, but a data object is stored only on one system. A first question to answer is how to distribute which data in which node? That depends on the application, for example a Youtube-like application could put italian video in italian server because they are mostly like to be requested there.

Data Fragment / Data Sharding

Each partition of data is called data fragment. Another term is data sharding, most common refering to horizontal partitioning of data before distributed systems.

Two other options for distributing data includes:

  • Horizontal partitioning (same as Range Partitions, on a different node): works well only if partitions are balanced
  • Vertical Partitioning, column wise partition: it make sense where you have some server that makes queries on attribute and some other server that make queries on different attributes.

Contrary to the previous scenario, here we have a very fast writing efficiency because we only update data on a single node. Also lock manager and transaction manager have an easier job to do here.

In real life case use, you are between these two extremes.

Data Consistency or Synchronization of data

Data consistency means the coherence of all the copies. This is a very strong requirement for big data applications, and many NOSQL systems doesn’t satisfy this. It depends on the real case application.

Example

Say you have a booking application for ticket in a plane. If one node goes down when you are booking the ticket, what happens? If you want to be strictly consistent, then you have to wait up until that node is up again, then synchronize and only after you can commit the fact that you booked the plane ticket.

This example for saying that distributed systems generally don’t scale very well

Scaling in distributed systems

There are two ways to scale:

  • Scale Up: improve hardware. The problem is that the price goes up until you have a technological limit but the price grows exponential. You get a very small amount of performance that costs lot of money. And you cannot go over the state of the art.
    • Cons: the price grows exponentially as you buy the best hardware, but there is a technological limit given by the state of art of hardware.
  • Scale out: buy more nodes (:=servers) and put them in a cluster. So with scale out, the more i pay, the more i get more computers. So, in theory i have a linear relation between price and performance. In reality is sublinear why i have the overhead to manage distribution and comunication
    • Pro: The more i pay, the more the performance increase, theoretically to infinite.
    • Cons: The relation between price and performance is sublinear because with more nodes i have overhead to manage distribution and comunication. (parallelism)

In general, data consistency is the strongest constrain you can put on a concurrent system because concurrency systems do not scale well. I.e at some point you get so much overhead because of synchronization of data, that scaling out is not so much convenient and you need to scale up.

NOSQL systems generally scale better than relational systems: they give up on the consistency to gain in term of performance.

Cap Theorem

  • Chapter 24 | Fundamentals of Database Systems | Navathi etc

  • C=Consistency, coherence of copies

  • A=Availability

  • P=Partition Tolerance, means robustness to failure of nodes.

https://external-content.duckduckgo.com/iu/?u=https%3A%2F%2Fwww.nitendratech.com%2Fwp-content%2Fuploads%2F2020%2F01%2Fcap_theorem.png&f=1&nofb=1&ipt=f7558e8b420ad2f6c3b4bc28ebd556cf97fe2b5ec1547fb5e6bdedb525af0349&ipo=images

The CAP Theorem is central to NoSQL systems, stating that in a distributed system, you cannot simultaneously achieve all three characteristics: consistency, availability, and partition tolerance. Pick two.

For example, if you have consistency and availability, your data is reachable and consistent, but this typically means zero replication. As soon as one node goes down, you lose partition tolerance. For example, If you have consistency and partition tolerance, you have replication, but you cannot have availability because you need to lock the data to update it. In both cases, the system does not scale effectively.

When you abandon SQL you want to prioritize performance but quit absolute consistency constraint. So it’s up to the programmer to solve the system incoherence.

Rule of thumbs

How to solve this issue? There are some suggestions

  1. Do the transactions as shorts as possible. Longer transactions leads to bigger problems.
  2. Establish policies to manage incoherence.

Example

Say you want to book a room, you design a system for booking so you have the rooms. How do you manage the transaction when to book the room for a client? There are different options:

  • FIFO: As soon as the client put the item in the chart
  • On paying: works but you could have two clients that buy the same room at the same time and only one will suceed.
  • Temporarly “lock” of the item with a countdown: like in Trenitalia

Note: this is not a technical problem

There are different technical options but no optimal, it’s a business problem

Other companies do overbooking, using statistics they know that say 1% of people miss the airplane, so they can sell 1% of more places on the plane.

Sometime, on rooms booking there is also the warning “the room may not be available” because they do not lock for 10 minutes. Some other systems do not have the “cart”, you select it and just go pay it.

Sometimes it happens that maybe there is a bonus. I give a bonus to a people that buy a new bike. What is the criteria? FIFO? First In First Out? It’s crazy, it’s clear that you will have milions of requests and in microseconds you will see who get the bonus.

For example you can do things like shard data by region, or by surname or so on. So when you are close to the deadline you get a pick and the systems goes down. Very often the criteria is reasonable, to make a system sized enough for manage data.

Data sharding by region

What should we know?

  • The volume of data
  • Number of requests We should have an idea, otherwise we can’t divide the data. Say we have 1 TB and 10.000 requests for day, each day i move 3TB of data of traffic.
  1. In House Solution: you buy the server, you deploy the server, put the infrastructure. It will gives you in terms of time and costs, (diagramma cartesiano). I spend a lot of money at the beginning and then zero for a certain amount of time, maintenance so some costs. Scale Up when i can’t manage more requests. You spend a lot of money, you exploit you systems and at certain points you have to scale
  2. Buy a system on cloud: you are renting a system, so your rate of rental could be even costant, less than the cost of the server. And you could have a price increase somewhere.

Cloud vs in-house

The total cost of data manging, the integral of the graph, your in house solution will be a step-wise function. Your cloud will be a linearly function that over time grows. At some point there will be a breakpoint: from now on, you spend more on cloud than by having the server. So, at the beginning the cloud solution is very cheap, but you have to sum the rental rates over the months, the years, so it can well happen that a certain point the cloud will be.

https://external-content.duckduckgo.com/iu/?u=https%3A%2F%2Fhtml.scirp.org%2Ffile%2F7-2121263x6.png&f=1&nofb=1&ipt=9e3d9387fcb7399282e8d51499aa452f84771a7ead73e04ae1c4a7e2a92cca16&ipo=images

You can buy relational systems on the cloud, but you need to consider whether it’s convenient and if you can handle potential inconsistencies in your data. For instance, selling tickets requires consistency. Why should I lock the tickets? If I used the Trenitalia solution to store 100 tickets, it would take more than 10 minutes to sell them all, whereas I could sell them in one second. Naturally, this second solution would upset many customers.

Another reason to maintain an in-house system is to meet legal privacy requirements. You may have sensitive data that needs to be stored in a separate database.

Lezione 12

ACID vs Base Properties

Let’s recall the properties of ACID transactions and BASE systems.

ACID Properties

  • Atomicity: Ensures that multiple operations within a transaction are treated as a single unit. Either all operations succeed, or none do, with no intermediate states.
  • Consistency: Guarantees that a transaction brings the database from one valid state to another, maintaining database rules and integrity constraints.
  • Isolation: Ensures that transactions are executed independently, without interference, so that intermediate results are not visible to other transactions.
  • Durability: Once a transaction is committed, its results are permanent and survive subsequent failures.

Imagine a time line. If at data is coherent and that i have multiple transactions: it’s like a walk from a coherent state to another coherent state. Incoherence is admitted only while the transaction operation is in course (temporary incoherence). But you must have absolute coherence at the start and at the end of the transaction.

Isolation is important because it means that multiple transaction should be executed as if they were isolated. It means that they execution is independent. Even if they work with the same data. The change should be persistence, that is guaranteed to a transaction system.

To have the garancy that a change is persistent, you need to stop clients from reading data and this slow down the system, possibly causing an overhead if you have lot of requests. This is why relational systems do not scale well. (See scaling in distributed systems)

BASE Properties

  • Basically Available: The system guarantees availability, even in the presence of partial system failures.
  • Soft state: The system’s state may change over time, even without input, due to eventual consistency.
  • Eventual consistency: The system will become consistent over time, ensuring that, given enough time, all replicas will converge to the same state.

The idea behind a transaction is to perform multiple operations atomically, ensuring that they either all succeed or all fail, with no intermediate states.

Conflicting operations

  • Multiple operations on the same data with atleast one writing: multiple users access the same data and do atleast one writing.

Example one

Consider the bank account transfer case from the previous lesson: the same bank account is shared between two people. They withdraw and deposit money simultaneously.

  • First operation: P1 reads the account balance (R(x)), which is €1000.
  • At : P1 withdraws €100, but P2 still reads the balance as €1000.
  • At ​: P2 writes the new balance as €1000 + €100.

Without control, P2 sees an extra €100, leading to a concurrency problem. This should not happen. A simple solution is to serialize operations in a queue, eliminating concurrency issues. However, this is impractical. Instead, concurrency protocols are used to address such problems

Example two

Dirty read: read of a value that has been modified by another transaction before the transaction is confirmed. Is dirty because you don’t know if the transaction that modify that value ended up with a success or fail so you read a value that is uncertain. So this is a problem.

Another potential problem could be:

  • P1 reads €1000 at .
  • P2 withdraws €100 at ​, reducing the balance to €900.
  • P2 delays the operation and reads €900, which seems correct.

But if the transaction fails, a rollback is needed. If the transaction is properly implemented, the €100 should be available again. The problem is that P2 read an incorrect balance, leading to a “dirty read.”

Dirty read

This problem occurs when a value modified by another transaction is read before that transaction is confirmed. It’s “dirty” because the transaction that modified the value may not succeed, making the read value uncertain.

Others known problems in multi client systems are phantoms, non repeatable reads, incorrect sum and so on.

In the bank account showcase example, using a relational system is acceptable because you will rarely have conflicting operations and locking data is acceptable.

Soft state

Means you allow dirty reads: you show to the client data that is being modified and is not necessary true.

Example - Cassandra and ACID

A NoSQL database like Cassandra can be ACID compliant at the tuple level. For instance, when inserting 500 tuples in a transaction, Cassandra treats each tuple insertion as a separate transaction. So it makes 500 transaction for 500 tuples.

Distributed system and long tail problem

Let’s talk about distributed system and horizontal scaling. Say i have an ecommorce that i manage multiple orders. Say i sell 100 computers, do i really need to know if there are exactly 98 or 97 computers? Can i tolerate overselling like in our previous example? This depends from resupplying. If i can resupply fastyle, i have no problem if people overbook because they would probably barely notice.

Long Tail Problem

In many businesses, the sales of goods typically follow a pattern resembling a Gaussian distribution. The majority of customers tend to purchase the same popular items, which require fast restocking. About 20% of the goods account for 80% of the requests, while the remaining 80% of goods receive only 20% of the requests.

Ofcourse in real life you don’t know what are you gonna sell more in advance, and you must use historically data and try to guess and then do your shelf. In some business are you get a long tail: the requests are more spread. If you have a long tail maybe is better to have a shelf that refill these goods.

Alternatively, you could operate with zero inventory and purchase goods after receiving customer orders. In this model, there’s no need for predictive forecasting; you receive payment upfront and procure items from your supplier afterward. This approach relies on short supply times, enabling infinite availability with zero upfront investment. However, it’s crucial to consider supply chain disruptions, as even large companies, like car manufacturers, face crises due to their just-in-time production approach. This strategy focuses on producing only what is sold or guaranteed for sale.

In the first case you need the order to be coherent with the availability. In the second case there is no need for a relational system. I just store all the orders and requests, sequentially, ready many, never conficting operations you never overwrite. If it is not available i will buy from the supply.

In this example, one can naively think it is necessary to show availability to clients. It’s not, and it’s not stopping them to buy from you or satisfying them. In this case nosql is good. Another example is social media: there are no conficting operations, no one is going to overwrite you, it’s going to answer you, so no concurrency problem.

Versioning

Consider another scenario: multiple individuals editing the same document, which is replicated across a distributed system. While there’s technically only one document, there are multiple copies, leading to conflicting operations.

One way to manage this is through versioning. Multiple versions can exist simultaneously, allowing simultaneous edits. In some cases, such as with documents, merging versions may be feasible. However, in scenarios like GitHub, where code is involved, merging may not always be viable.

Versioning is another trick to skip the need for coherence found the relational system.

Versioning serves as a workaround to bypass the need for coherence found in relational systems.

Coherence with BLOBs in NoSQL

Say you are netflix or amazon prime video, your basic data requirement is to store big files. Once you update a movie in your catalogue that’s it you don’t modify it. Clients can only reads, there are no conficting operations. Nobody is going to write into a BLOB.

What it can be done is to use the locality principe to reduce network traffic. In this case full replication is good because we have no writing. If you use statistics you can put copies of most requested films in all the nodes.

Another thing is the speed at which you deliver the data: as most people see the film in streaming, they do not download it. So in streaming you have a certain volume of data but there is no point in going fast. People cannot see a film at 4x the speed, so once you get a certain volume of data: definition of film, capacity of the network, you have a finite amount of data to transfer. Because the consumer of data has a fixed velocity. So you have a buffer, you can preload it and make it available on all the clients. In this case you ust need a fast buffer and a lot of replication to have a performance system. There is no need of SQL.

Other types of distributed databases concept

  • Federated databases are databases that operate independently but are connected at the application level. They works kind of like a middleware linking various services and presenting a unified interface to clients. Each database retains autonomy, communicating through the middleware, which standardizes API connections to different servers. This setup offers a higher degree of autonomy compared to distributed databases.
  • Polyglot database: instead of relying solely on a single type of database (e.g., relational or NoSQL), a polyglot database environment leverages a mix of database technologies based on the specific needs of different parts of the application.
    • Polyglot are common because often core data is in relational form, but you also want part of that data in non relational system

Master slave example in relational e non-relational distributed database

Imagine a scenario with a master node and several slave nodes in a database setup. How can the slaves contribute to meeting client needs? By redirecting client requests to the slaves, we can speed up read operations, leveraging the distributed architecture. However, the master becomes a bottleneck, especially for write operations. When you write, you should first ask to the master where data is being written.

In a non-relationa master-slave setup, requests to modify data can be sent to the master without locking the data, allowing clients to access data without interruption. Eventually, synchronization occurs between the master and slaves, ensuring consistency.

If a node goes down, in a relational system, it remains inaccessible until synchronization is complete. Conversely, in a non-relational system, the node becomes functional once again, even during ongoing synchronization. It’s devolved to the application how to reconciliate the different values. Depends on the application.