CumuloNimbo:
A Cloud Scalable SQL Database
Rui Carlos Gonçalves
HASLab, INESC TEC & U. Minho
Porto Linux, 2016

RoadMap› Background Research› Scalable Transaction Processing› Scalable Analytics

2Research

Yousuf Ahmad

Ainhoa Azqueta

Ivan Brondino

Fábio Coelho

Francisco Cruz

Bettina Kemme

Rui Gonçalves

Ricardo Jimenez

Miguel Matos

Marta Patiño

José Pereira

Ricardo Vilaça

 These projects received funding from the European Union's Seventh Framework Programme for research, technologicaldevelopment and demonstration under grant agreements no 257993, 611068 and 619606.
 3

Rui Oliveira

Background› Big Data: Explosion of data being generated and stored

every day› Holds valuable knowledge for organizations' operation› Opportunity to improve efficiency› New challenges to store and process massive amountsof data4

Scalable Transaction Processing

5

Scalable Transaction Processing› To offer the classical ACID properties with useful isolationlevels, current clustered databases scale poorly both interms of performance and efficiency

5

Scalable Transaction Processing› To offer the classical ACID properties with useful isolationlevels, current clustered databases scale poorly both interms of performance and efficiency› To scale, current systems need to sacrifice atomicity,

consistency and overall ease of use

5

Scalable Transaction Processing› To offer the classical ACID properties with useful isolationlevels, current clustered databases scale poorly both interms of performance and efficiency› To scale, current systems need to sacrifice atomicity,

consistency and overall ease of use› The Big Data challenge to current clustered databases is

to allow analytical processing over the freshest data5

Scalable Transaction Processing› To offer the classical ACID properties with useful isolationlevels, current clustered databases scale poorly both interms of performance and efficiency› To scale, current systems need to sacrifice atomicity,

consistency and overall ease of use› The Big Data challenge to current clustered databases is

to allow analytical processing over the freshest data

› Requires Hybrid Transactional and Analytical Processing -simultaneous transactional and analytical workloads

5CumuloNimbo

6

CumuloNimbo› CumuloNimbo is a framework for multi-tier applications

that provides ultra-scalable and fault-tolerant processingof OLTP and OLAP workloads

6CumuloNimbo› CumuloNimbo is a framework for multi-tier applications

that provides ultra-scalable and fault-tolerant processingof OLTP and OLAP workloads› It provides a standard SQL interface and full ACID

transactional support without resorting to sharding

6

CumuloNimbo› CumuloNimbo is a framework for multi-tier applications

that provides ultra-scalable and fault-tolerant processingof OLTP and OLAP workloads› It provides a standard SQL interface and full ACID

transactional support without resorting to sharding› Data is persisted in a distributed fault-tolerant data store

6

CumuloNimbo› CumuloNimbo is a framework for multi-tier applications

that provides ultra-scalable and fault-tolerant processingof OLTP and OLAP workloads› It provides a standard SQL interface and full ACID

transactional support without resorting to sharding› Data is persisted in a distributed fault-tolerant data store› Scalability is achieved by distributing request execution

and transaction control across many compute nodes

6

Global Architecture
Architecture

ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

Elastic
ManagerCloud
Deployer
Platform
Management
Framework

The CumuloNimbo system consists of multiple layers each of them having a specificfunctionality. In order to achieve the desired throughput, each layer can be scaled
independently by adding more servers. Instances of different layers might be collocated to
improve response time performance.
7

Global Architecture
Architecture

ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

8

Elastic
ManagerCloud
Deployer
Platform
Management
Framework

The Query Engine layer, based on Apache Derby query engine, provides the standard
relational interface (SQL) including the transaction commands (commit, abort, and begin).
The query layer itself is responsible for planning, optimizing and executing queries.
However, it does not perform transaction management nor is it responsible for data
storage.

Global Architecture
Architecture

ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

Elastic
ManagerCloud
Deployer
Platform
Management
Framework

The NoSQL Data Store layer uses an advanced data store that is designed for inherent
scalability in terms of data size and processing requirements. Our system is based on
HBase, an elastic key-value store.
9

Global Architecture
Architecture

ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

Elastic
ManagerCloud
Deployer
Platform
Management
Framework

The NoSQL Data Store sits on top of the Hadoop Distributed File System that provides
persistent and fault-tolerant data storage. The NoSQL Data Store tables, as well as its
write-ahead log, are persisted in HDFS.
10Global Architecture
Architecture

ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

Elastic
ManagerCloud
Deployer
Platform
Management
Framework

The Transaction Management is handled in a holistic manner providing full ACID
properties across the entire stack.

11

Global Architecture
Architecture

ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

Elastic
ManagerCloud
Deployer
Platform
Management
Framework

The Platform Management Framework takes care of tasks such as deployment,
monitoring, dynamic load balancing and elasticity. Each instance on each layer has amonitor collecting data about resource usage and performance metrics that are reported
to a central monitor.
12

OLTP

13

Scalable Transaction Processing› Snapshot Isolation level› Each transaction reads the data from a snapshot that

represents all committed values at the start time of atransaction› Conflicts are detected on updates: Whenever two

transactions are concurrent (neither commits before theother starts), and they want to write a common data
item, one of them has to abort› Snapshot isolation requires a multi-version systemproviding snapshot read and snapshot write properties

14

Scalable Transaction Processing
Data Store

Query Engine

Txn Mng
TS=1

StartTxn
Write(x, 27)

Transaction execution
under snapshot isolation ina centralized system

x:= 27Read(x)Î27

Read(y)Commit
TS++
TS=2

15

Scalable Transaction Processing
Tx Management

Query&Engine› Naive scale up approach

Data&Store› Limitation 1: Serial and Atomic Commit Processing

Making commit processing appear as a single atomic
action that is spread across multiple nodes becomes
extremely expensive in terms of the time it takes tocomplete the action, as well as the negative impact it
has on concurrency

16

Scalable Transaction Processing
Tx Management

Query&Engine› Naive scale up approach

Data&Store› Limitation 2: Monolithic Transactional Processing

The global transaction management component caneasily become a bottleneck, performing many different
tasks and connecting to a large number of othercomponents.

17

Scalable Transaction Processing
Tx Management

Query&Engine› Naive scale up approach› Limitation 3: Synchronous Communication

Data&Store

Transaction execution is prolonged by including many
synchronous message exchanges. For each write
operation, the transaction first checks for conflicts at thetransaction manager, and then writes the changes to thedata store. Furthermore, the start and commit of thetransaction require additional message rounds with thetransaction manager and/or the data store.

18

Scalable Transaction Processing
Tx Management

Query&Engine› Naive scale up approach› Limitation 4: Message Overhead

Data&StoreAt large transaction rates all components have to handle
a large amount of messages which can quickly becomea major overhead, for instance, a round-trip message
per updated item.

19

The CumuloNimbo BreakthroughUntangle and scale out the ACID properties independently

20

The CumuloNimbo BreakthroughUntangle and scale out the ACID properties independently

21

The CumuloNimbo BreakthroughUntangle and scale out the ACID properties independently
Local Tx
Managers

Con ict
Managers

Snapshot Server
Loggers

Commit Sequencerfl21

Principles of the Approach› Decoupling Update Visibility and Atomic Commit› Proactive Timestamp Management› Parallelization and Distribution› Asynchronous Messaging and Batching22

Decoupling Update Visibility and Atomic Commit
Local0Tx Manager

Data0Store

StartTxn

Logger

G etS tartTS ()

Snapshot Commit
Server Sequencer

TS=1

1

Distinguish Snapshot
Timestamp from Commit
Timestamp.
Updates to the DataStore are now outside of
the response path.

Commit

GetCom m itTS( )TS++

2
Log0WriteS et

Reply0to
ClientDurableetWrite0WriteS
R eadable

23

TS=2

Decoupling Update Visibility and Atomic Commit
The Snapshot server keeps
track of the most recent
snapshot that is consistent.
Its TS should be such that there
is no previous Commit TS thata) is not yet readable or b) has
been discarded.
This way, transactions cancommit in parallel andconsistency be preserved.

24

Snapshot Server
Gets reports ofdurable &readable TSs

Getsreports of
discarded
TSs
Keeps track of andreports mostrecent consistentTSDecoupling Update Visibility and Atomic Commit
Coordination between Snapshot and Commit servers
Sequence of timestamps received by the Snapshot Server

11
T AM P
T IM ES

11

T AM P
T IM ES15

15

T AM P
T IM ES1212

T
T IM ES

4
AM P 1

14

T
T IM ES

3
AM P 1

13Time
Evolution of the current snapshot at the Snapshot Server

25

P 11
STAM
TIME

11
AMP
T
S
E
TIM

12
AMP
T
S
E
TIM

P 12
TAM
S
E
M
TI

P 15
TAM
S
E
M
TI

11

11

12

12

15

Proactive Timestamp Management
Local0Tx Manager

Data0Store

StartTxn

Proactive timestamp serving
to the rescue

G etS tartTS (

)

TS=1

Commit

GetCom m itTS( )TS++

2
Log0WriteS et

Reply0to
ClientDurableetWrite0WriteS
R eadable

26

Snapshot Commit
Server Sequencer1

Two independent timestamp
services
Timestamp management canbecome ultimate bottlenecks

Logger

TS=2

Proactive Timestamp Management
Commit Sequencer
Once a transaction enters thecommit phase it is guaranteed that it
does not conflict with any othertransaction

Local*Txn Mng
B1

0*1* 2T1

B2 * 5*
3*4

The commit timestamp is only
T1
T2
T3
needed to tag the updates with thatB3 * *
6*7
0 1 X
2 3 4 5commit timestamp

27

10ms
3

T2

0 1 2

It does not matter which timestamp
each transaction gets!

Commit*Server

10ms
2

10ms
1

Parallelization and Distribution
All CumuloNimbo architecture is distributed and mostcomponents can be run in parallel

G etS tartTS

Dura

m itTS )
m
o
C
(
e
l
dab
ble&R ea

Write(data,0Com m itTS

ad
(S

mm
tCo
itT

Re

Ge

Conflict0
Conflict0
Conflict0
Mng
Mng
Mng

ta
rtT
S

)

Snapshot0Server

Query
Query
Query
EngineEngineEngine+++?
Local0ict
Local0Confl
Local0TM
TM
TM
Write(data)

Log
Dura

R eadableble

Snapshot0ServerStartTS

Data
Data
Store
) DataStore
Store

LoggerLoggerLogger

S

Data
Data
Data
Store
Store
Store
Transaction0
Start

Transaction0
Execution

Commit0ServerTransaction0 Make0Updates0 Make0Updates0 Report0Readability Report0
Serialization Durable0(log)Readable&0Durability
Visibility
Transaction0Commit

28Parallelization and Distribution
Local Transaction Managers
Manage the transactions life cycle
Can have parallel instances
Collocated with Query Engines

G etS tartTS

Dura

m itTS )
m
o
C
(
e
l
dab
ble&R ea

Write(data,0Com m itTS

ad
(S

mm
tCo
itT

Re

Ge

Conflict0
Conflict0
Conflict0
Mng
Mng
Mng

ta
rtT
S

)

Snapshot0Server

Query
Query
Query
EngineEngineEngine+++?
Local0ict
Local0Confl
Local0TM
TM
TM
Write(data)

Log
Dura

R eadableble

Snapshot0ServerStartTS

Data
Data
Store
) DataStore
Store

LoggerLoggerLogger

S

Data
Data
Data
Store
Store
Store
Transaction0
Start

Transaction0
Execution

Commit0ServerTransaction0 Make0Updates0 Make0Updates0 Report0Readability Report0
Serialization Durable0(log)Readable&0Durability
Visibility
Transaction0Commit

29Parallelization and Distribution
Conflict Managers
Each CM handles a disjoint subset of data record keys
Can have parallel instances

G etS tartTS

Dura

m itTS )
m
o
C
(
e
l
dab
ble&R ea

Write(data,0Com m itTS

ad
(S

mm
tCo
itT

Re

Ge

Conflict0
Conflict0
Conflict0
Mng
Mng
Mng

ta
rtT
S

)

Snapshot0Server

Query
Query
Query
EngineEngineEngine+++?
Local0ict
Local0Confl
Local0TM
TM
TM
Write(data)

Log
Dura

R eadableble

Snapshot0ServerStartTS

Data
Data
Store
) DataStore
Store

LoggerLoggerLogger

S

Data
Data
Data
Store
Store
Store
Transaction0
Start

Transaction0
Execution

Commit0ServerTransaction0 Make0Updates0 Make0Updates0 Report0Readability Report0
Serialization Durable0(log)Readable&0Durability
Visibility
Transaction0Commit

30Parallelization and Distribution
Loggers
Each Logger handles a fraction of the log records
Can have parallel and replicated instances
Independent; do not coordinate with each otherG etS tartTS

Dura

m itTS )
m
o
C
(
e
l
dab
ble&R ea

Write(data,0Com m itTS

ad
(S

mm
tCo
itT

Re

Ge

Conflict0
Conflict0
Conflict0
Mng
Mng
Mng

ta
rtT
S

)

Snapshot0Server

Query
Query
Query
EngineEngineEngine+++?
Local0ict
Local0Confl
Local0TM
TM
TM
Write(data)

Log
Dura

R eadableble

Snapshot0ServerStartTS

Data
Data
Store
) DataStore
Store

LoggerLoggerLogger

S

Data
Data
Data
Store
Store
Store
Transaction0
Start

Transaction0
Execution

Commit0ServerTransaction0 Make0Updates0 Make0Updates0 Report0Readability Report0
Serialization Durable0(log)Readable&0Durability
Visibility
Transaction0Commit

31Asynchronous Messaging and Batching› Conflict Managers› Conflict checks are asynchronous: non-conflicting

transactions run faster vs. abort detection is delayed› Conflict checks are batched› On commit, outstanding conflict checks are synchronized› Update propagation› While read operations are synchronous, updates are

made asynchronously and batched per data store

32Asynchronous Messaging and Batching› The local transaction managers report periodically about

the number of committed update transactions per second› The commit sequencer distributes batches of commit

timestamps to the local transaction managers› The snapshot server gets periodically batches oftimestamps (both used and discarded) from local
transaction managers› The snapshot server reports periodically to local

transaction managers the most current consistentsnapshot

33

OLAP

34

Scalable Analytics
Current Data Warehousing solutions rely in ETLs to update
data.
Transactional DB
Consistent Updates

OLTP

Data Warehouse
Fast Analytical Queries

Copy Data (ETL)OLAP

Cost of ETLs 80% of Business Analytics !
The Big Data challenge to current clustered databases is toallow analytical processing over the freshest data.
35

Scalable Analytics
Analytical queries over the transactional DB:
Real-Time Analytics (No ETLs)
Transactional DB
Consistent Updates

Data Warehouse
Fast Analytical Queries

OLTP

OLAP

HTAP -80% cost in Business Analytics !

36

OLTPGlobal
Architecture
Architecture
ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

37

Elastic
ManagerCloud
Deployer
Platform
Management
Framework

OLTPGlobal
Architecture
Architecture
ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

37

Elastic
ManagerCloud
Deployer
Platform
Management
Framework

OLTPGlobal
Architecture
Architecture
ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

37

Elastic
ManagerCloud
Deployer
Platform
Management
FrameworkOLAPGlobal
Architecture
Architecture
ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

38

Elastic
ManagerCloud
Deployer
Platform
Management
FrameworkOLAPGlobal
Architecture
Architecture
ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

38

Elastic
ManagerCloud
Deployer
Platform
Management
FrameworkOLAPGlobal
Architecture
Architecture
ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

38

Elastic
ManagerCloud
Deployer
Platform
Management
FrameworkOLAPGlobal
Architecture
Architecture
ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

38

Elastic
ManagerCloud
Deployer
Platform
Management
FrameworkOLAPGlobal
Architecture
Architecture
ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

Elastic
ManagerCloud
Deployer
Platform
Management
Framework

CumuloNimbo can leverage data locality, when query engines, HBase and HDFS are colocated on the same physical machine.
38OLAPGlobal
Architecture
Architecture
ConflictManagers

Local Txn Mngs

CommitSequencer
SnapshotServer

Application*Server
Application Server(JBoss+Hibernate)

Query Engine (Derby)

Monitors

NO SQL Data Store (Hbase)
NoSQL&Data&Store&(HBase)

Load
Balancers

Distributed File System (HDFS)
Loggers
Holistic
Transaction
Management

Elastic
ManagerCloud
Deployer
Platform
Management
Framework

CumuloNimbo can leverage data locality, when query engines, HBase and HDFS are colocated on the same physical machine.
38Parallel Coordination› Maintains a set of symmetric workers for a user connection› Initializes a communication middleware for efficient intra-

query row exchange among workers› Spawns a parallel query plan for all workers to execute› Schedules disjoint subsets of data to scan to the different

workers› Collects results and errors

39

Parallel Query Plans
Project
Restrict

Grouped
Agg.

Project
RestrictScan

40

stateless operator

stateful operator

select a, avg(b) from t
where c > 0group by ahaving avg(b) > 0

Parallel Query Plans
Project
Restrict

Project
Restrict

GroupedAgg.

Grouped
Agg.

Project
Restrict

Project
Restrict

Shard
Scan

41

partial (incorrect)6resultsfor6all groups

each6worker
scans6a6shard

Shard
Scan

Parallel Query Plans
ShuffleTarget0

Project
Restrict

Grouped
Agg.

Shard
Scan

:

:

42

Project
Restrictfull8results8forpart8of8the8groups

ShuffleProject
Restrict

ShuffleGroupedAgg.

ShuffleTargethash(a) mod #workers

Project
Restrict

Shard
Scan

Communication Middleware› Provides efficient row exchange for shuffle operators› Supports hash-based shuffling› Asynchronous and multi-stage› Multiple implementations available (e.g. sockets, RDMA

Verbs)

43

Conclusions› CumuloNimbo platform offers distributed and highly

scalable SQL processing, with full ACID properties› OLTP components iterated over 5 years: beta ready› OLAP has a fully working implementation› Ongoing work
› XA
› HTAP (OLAP+OLTP)

44

Thank you for your attention

45