Aaron Morton

PRIMARY KEY's in CQL

2013-01-11T00:00:00-08:00

The final version of CQL 3 that ships with Cassandra v1.2 adds some new features to the PRIMARY KEY clause. It overloads the concept in ways that differ from the standard SQL definition, and in some places shares ideas with Hive. But from a Cassandra point of view it allows for the same flexibility as the Thrift API.

Schema, Schema, Every Where

There are two ways to specify the primary key in the CREATE TABLE statement. It can be specified in line.

create table foo (
    bar     int PRIMARY KEY, 
    baz     int
);

Or it can be specified as a separate clause, which is the method we will be using.

create table foo (
    bar     int, 
    baz     int,
    PRIMARY KEY (bar)
);

The definition of the PRIMARY KEY clause in the spec can appear confusing at first.

PRIMARY KEY '(' <partition-key> ( ',' <identifier> )* ')'

However the comments further down the tell us all we need to know.

In CQL, the order in which columns are defined for the PRIMARY KEY matters. The first column of the key is called the partition key. It has the property that all the rows sharing the same partition key (even across table in fact) are stored on the same physical node. Also, insertion/update/deletion on rows sharing the same partition key for a given table are performed atomically and in isolation. Note that it is possible to have a composite partition key, i.e. a partition key formed of multiple columns, using an extra set of parentheses to define which columns forms the partition key.

So lets get started.

The Setup

In the examples below I used a 3 node local cluster created with the ccm tool from Sylvain Lebresne. Before starting the cluster I brought up 2 additional network interfaces for the nodes to bind to.

$ sudo ifconfig lo0 alias 127.0.0.2 up
$ sudo ifconfig lo0 alias 127.0.0.3 up

I could then start a 3 node cluster using version 1.2.0 from the ccm directory.

$ ccm create dev -v 1.2.0
Current cluster is now: dev
$ ccm populate -n 3
$ ccm start

Checked everything was as expected.

$ ccm node1 ring
Note: Ownership information does not include topology; for complete information, specify a keyspace

Datacenter: datacenter1
==========
Address         Rack        Status State   Load            Owns                Token                                       
                                                                               3074457345618258602                         
127.0.0.1       rack1       Up     Normal  24.88 KB        33.33%              -9223372036854775808                        
127.0.0.2       rack1       Up     Normal  24.89 KB        33.33%              -3074457345618258603                        
127.0.0.3       rack1       Up     Normal  15.54 KB        33.33%              3074457345618258602

Then I started a cqlsh (in bin/ of the standard distribution) session against node 1.

$ bin/cqlsh 127.0.0.1
Connected to dev at 127.0.0.1:9160.
[cqlsh 2.3.0 | Cassandra 1.2.0-SNAPSHOT | CQL spec 3.0.0 | Thrift protocol 19.35.0]
Use HELP for help.
cqlsh>

And created a Keyspace with RF 1.

cqlsh> create keyspace dev 
   ... WITH replication = {'class':'SimpleStrategy', 'replication_factor':1};
cqlsh> use dev;
cqlsh:dev>

The Sound of One Column Indexing

Cassandra 1.2 allows tables to be defined with one column that is also the PRIMARY KEY. If you’ve used Cassandra before this may sound muy loco as internally a row without columns is purged during compaction. This allows rows that only contain ExpiringColumns to be automatically removed. If you wanted a row without any columns you would need a place holder column, and this pretty much what CQL 3 does.

My one column table looked like this.

CREATE TABLE device (
  device_id int,
  PRIMARY KEY (device_id)
);

I then put three rows in it.

INSERT INTO device 
(device_id)
values
(1);
INSERT INTO device 
(device_id)
values
(2);
INSERT INTO device 
(device_id)
values
(3);

In cqlsh this all looked sane.

cqlsh:dev> select * from device;

 device_id
-----------
         1
         2
         3

Jumping over to the old fashioned cassandra-cli we can see the place holder columns.

$ bin/cassandra-cli -h 127.0.0.1
Connected to: "dev" on 127.0.0.1/9160
Welcome to Cassandra CLI version 1.2.0

Type 'help;' or '?' for help.
Type 'quit;' or 'exit;' to quit.

[default@unknown] use dev;
Authenticated to keyspace: dev
[default@dev] list device;
Using default limit of 100
Using default column limit of 100
-------------------
RowKey: 1
=> (column=, value=, timestamp=1357864824406000)
-------------------
RowKey: 2
=> (column=, value=, timestamp=1357864824413000)
-------------------
RowKey: 3
=> (column=, value=, timestamp=1357864825075000)

3 Rows Returned.
Elapsed time: 49 msec(s).

Partitioning and Clustering

The PRIMARY KEY definition is made up of two parts: the Partition Key and the Clustering Columns. The first part maps to the storage engine row key, while the second is used to group columns in a row. In the storage engine the columns are grouped by prefixing their name with the value of the clustering columns. This is a standard design pattern when using the Thrift API. But now CQL takes care of transposing the clustering column values to and from the non key fields in the table.

My table with both a partitioning key and clustering columns looked like this.

CREATE TABLE device_check (
  device_id   int,
  checked_at  timestamp, 
  is_power    boolean, 
  is_locked   boolean,
  PRIMARY KEY (device_id, checked_at)
);

The partitioning key is the device_id and the clustering column is checked_at. The specification allows for more than one clustering column, I just chose one here. It also allows for multiple partitioning key columns as we will see later.

To see what these keys do I inserted some data. As before there are three devices in the example and each one is checked once a month to see if it is locked and powered.

INSERT INTO device_check
  (device_id, checked_at, is_power, is_locked)
values
  (1, '2013-01-01T09:00+1300', true, true)
;
INSERT INTO device_check
  (device_id, checked_at, is_power, is_locked)
values
  (2, '2013-01-01T09:10+1300', true, true)
;
INSERT INTO device_check
  (device_id, checked_at, is_power, is_locked)
values
  (3, '2013-01-01T09:10+1300', true, false)
;
INSERT INTO device_check
  (device_id, checked_at, is_power, is_locked)
values
  (1, '2013-02-01T09:00+1300', true, false)
;
INSERT INTO device_check
  (device_id, checked_at, is_power, is_locked)
values
  (2, '2013-02-01T09:10+1300', true, false)
;
INSERT INTO device_check
  (device_id, checked_at, is_power, is_locked)
values
  (3, '2013-02-01T09:10+1300', true, true)
;

Poking around in cqlsh everything looks as expected.

cqlsh:dev> select * from device_check;

 device_id | checked_at               | is_locked | is_power
-----------+--------------------------+-----------+----------
         1 | 2013-01-01 09:00:00+1300 |      True |     True
         1 | 2013-02-01 09:00:00+1300 |     False |     True
         2 | 2013-01-01 09:10:00+1300 |      True |     True
         2 | 2013-02-01 09:10:00+1300 |     False |     True
         3 | 2013-01-01 09:10:00+1300 |     False |     True
         3 | 2013-02-01 09:10:00+1300 |      True |     True
cqlsh:dev> select * from device_check where device_id = 1;

 device_id | checked_at               | is_locked | is_power
-----------+--------------------------+-----------+----------
         1 | 2013-01-01 09:00:00+1300 |      True |     True
         1 | 2013-02-01 09:00:00+1300 |     False |     True

So back to the cassandra-cli we go to see what’s happening with the clustering columns for device 1.

[default@dev] get device_check[1];
=> (column=2013-01-01 09\:00\:00+1300:, value=, timestamp=1357866010549000)
=> (column=2013-01-01 09\:00\:00+1300:is_locked, value=01, timestamp=1357866010549000)
=> (column=2013-01-01 09\:00\:00+1300:is_power, value=01, timestamp=1357866010549000)
=> (column=2013-02-01 09\:00\:00+1300:, value=, timestamp=1357866056217000)
=> (column=2013-02-01 09\:00\:00+1300:is_locked, value=00, timestamp=1357866056217000)
=> (column=2013-02-01 09\:00\:00+1300:is_power, value=01, timestamp=1357866056217000)
Returned 6 results.
Elapsed time: 38 msec(s).

Where we had two rows in cqlsh we now have one row with six columns in cassandra-cli which uses the Thrift API. CQL is mapping multiple instances of our entity (the device_check) to the same partition, and the partition is identified by the value of device_id. And you can probably guess that the partition is implemented as a row in the storage engine. The column names look a little strange, and technically I should not call them columns. In the current Cassandra lexicon the internal storage engine columns are called Cells, Columns are used for CQL. The first cell / storage column has the value of the first checked_at CQL column as it’s name 2013-01-01 09:00:00+1300:. The ’:’ indicates there are multiple components to this cell name, however this cell does not supply values for all parts. The cell does not have a value as the CQL column value is stored in the cell name. The second cell has a value for checked_at and the name of the first none primary key column is_locked 2013-01-01 09:00:00+1300:islocked_. In this case the cell value is the value for the is_locked CQL column. This pattern continues for the second CQL row, with checked_at equal to 2013-02-01 09:00:00+1300, and it will continue for all entities in this partition.

Now to take a look at the effect of the partition key on data placement. My expectation is that each unique device_id value, and so each partition, will be stored on a different node. In reality the storage engine rows are randomly distributed between nodes so I’ve adjusted my expectations appropriately. I also expect that each row will be replicated once, as I set the replication_factor to one.

To see the node a row is stored on we use the nodetool getendpoints command. It returns the replicas for a (storage engine) row key in a given Keyspace and Column Family.

$ bin/nodetool -h 127.0.0.1 -p 7100 getendpoints dev device_check 1
127.0.0.2
$ bin/nodetool -h 127.0.0.1 -p 7100 getendpoints dev device_check 2
127.0.0.2
$ bin/nodetool -h 127.0.0.1 -p 7100 getendpoints dev device_check 3
127.0.0.1

That’s close enough for me. Different partitions, identified by different device_id values, are stored on different nodes.

The Composite Enchilada

Now lets expand the partition key to use a composite type. This is useful when you have a time series and you need to partition the events to avoid huge rows. Rather than have one partition based on, in my example, the device_id I can have several where device_id is a part of the partition selection. Meaning CQL rows in this table with the same device_id may be located on different nodes. But all rows with the same values for the partitioning keys will be located on the same nodes.

For my example I used the devices I was tracking to check for dam dirty apes.

CREATE TABLE events (
  device_id   int,
  year_month  int,
  sequence    timestamp,
  pressure    int,
  temperature int,
  is_dam_dirty_apes  boolean,
  PRIMARY KEY ((device_id, year_month), sequence)
);

The partition keys are device_id and year_month, every event from the same device in the same month will be placed in the same partition. The grouping column is the time of the event which I called sequence (to avoid confusion with Cassandra timestamps).

So we turn on the network and start checking for apes.

insert into events 
(device_id, year_month, sequence, pressure, temperature, is_dam_dirty_apes)
values
(1, 201301, '2013-01-20T10:58:35+1300', 123, 10, false);
insert into events 
(device_id, year_month, sequence, pressure, temperature, is_dam_dirty_apes)
values
(2, 201301, '2013-01-20T10:58:40+1300', 456, 20, false);
insert into events 
(device_id, year_month, sequence, pressure, temperature, is_dam_dirty_apes)
values
(3, 201301, '2013-01-20T10:58:45+1300', 789, 30, true);
insert into events 
(device_id, year_month, sequence, pressure, temperature, is_dam_dirty_apes)
values
(1, 201302, '2013-02-20T10:58:35+1300', 1230, 11, true);
insert into events 
(device_id, year_month, sequence, pressure, temperature, is_dam_dirty_apes)
values
(2, 201302, '2013-02-20T10:58:40+1300', 4560, 21, true);
insert into events 
(device_id, year_month, sequence, pressure, temperature, is_dam_dirty_apes)
values
(3, 201302, '2013-02-20T10:58:45+1300', 7890, 31, true);

In cqlsh we now have 6 rows.

cqlsh:dev> select * from events;

 device_id | year_month | sequence                 | is_dam_dirty_apes | pressure | temperature
-----------+------------+--------------------------+-------------------+----------+-------------
         2 |     201302 | 2013-02-20 10:58:40+1300 |              True |     4560 |          21
         3 |     201302 | 2013-02-20 10:58:45+1300 |              True |     7890 |          31
         1 |     201302 | 2013-02-20 10:58:35+1300 |              True |     1230 |          11
         1 |     201301 | 2013-01-20 10:58:35+1300 |             False |      123 |          10
         3 |     201301 | 2013-01-20 10:58:45+1300 |              True |      789 |          30
         2 |     201301 | 2013-01-20 10:58:40+1300 |             False |      456 |          20

Checking in the cassandra-cli we see a similar layout to before, with the addition of a composite value used for the row key. Each CQL row has been transposed to four columns in a storage engine row.

[default@dev] list events;
Using default limit of 100
Using default column limit of 100
-------------------
RowKey: 2:201302
=> (column=2013-02-20 10\:58\:40+1300:, value=, timestamp=1357869160739000)
=> (column=2013-02-20 10\:58\:40+1300:is_dam_dirty_apes, value=01, timestamp=1357869160739000)
=> (column=2013-02-20 10\:58\:40+1300:pressure, value=000011d0, timestamp=1357869160739000)
=> (column=2013-02-20 10\:58\:40+1300:temperature, value=00000015, timestamp=1357869160739000)
-------------------
RowKey: 3:201302
=> (column=2013-02-20 10\:58\:45+1300:, value=, timestamp=1357869161380000)
=> (column=2013-02-20 10\:58\:45+1300:is_dam_dirty_apes, value=01, timestamp=1357869161380000)
=> (column=2013-02-20 10\:58\:45+1300:pressure, value=00001ed2, timestamp=1357869161380000)
=> (column=2013-02-20 10\:58\:45+1300:temperature, value=0000001f, timestamp=1357869161380000)
...

Nothing to scary there. Now over to nodetool to see where the rows are placed. We need to specify the value of device_id and year_month as these are used in the storage engine row key.

$ bin/nodetool -h 127.0.0.1 -p 7100 getendpoints dev events 1:201301
127.0.0.3
$ bin/nodetool -h 127.0.0.1 -p 7100 getendpoints dev events 2:201301
127.0.0.1
$ bin/nodetool -h 127.0.0.1 -p 7100 getendpoints dev events 3:201301
127.0.0.1
$ bin/nodetool -h 127.0.0.1 -p 7100 getendpoints dev events 1:201302
127.0.0.3
$ bin/nodetool -h 127.0.0.1 -p 7100 getendpoints dev events 2:201302
127.0.0.2
$ bin/nodetool -h 127.0.0.1 -p 7100 getendpoints dev events 3:201302
127.0.0.2

The two partitions each for devices 2 and 3 have been placed on different nodes. The partitions for device 1 are on the same node, but with enough nodes they would probably be on different ones.

Sorting Lists For Humans

2012-08-18T00:00:00-07:00

I like humans. Many of my friends a humans; my wife is a human. And I admire their ability to arbitrarily order items in a list. Applying a manual order to a list of items has been discussed a few times on the Cassandra user list. And I’ve been thinking about it recently.

To keep things simple let’s start with an artificially constrained problem. Our list will only contain 50 “things” that must be manually sorted. The things consist of an Integer id and an ASCII label. We can also say that the rate of changes is not excessive. In this scenario there are two issues we want to look at:

How do I move an item without re-ordering everything?
How do I handle concurrent changes to the list?

The Natural Order

My first instinct is to duplicate the list order in Cassandra Columns. For example we could use a Composite Column name such as (weight, id). Cassandra would then sort the Columns by weight, using id as a tie breaker for duplicate weight’s. To get the list in order we simply select the Columns from the row.

Moving items using this data model can be painful. Consider the (weight, id) list below:

(1, 101)
(2, 102)
(3, 103)
(4, 104)
(5, 105)

If we want to move item 105 to be before item 102 we would change it’s weight to 2. To avoid duplicate weights we would then change the weight of 3 and the items that follow it. The reason to avoid duplicate weight’s is that the sort order for duplicates would be based on the id. If we had a better tie breaker duplicate weights would not be a problem.

The unceasing March Of Time is a handy tie breaker. With it we can say that if two items have the same weight the most recently updated one should be ordered last. If there are items with duplicate (weight, timestamp) values we can then use the id to break the tie. To avoid confusion with the Timestamp Cassandra stores for Columns I’ll refer to our timestamp as the ‘sequence’ or ‘seq’.

Using the weights and sequence model the Column names wil have the form (weight, seq, id). If we initialise the weight and seq to the current time stamp our example from above may now look like:

(1, 1, 101)
(2, 2, 102)
(3, 3, 103)
(4, 4, 104)
(5, 5, 105)

To move an item so that it preceeds another we:

Set the weight to the moving item of the weight of the new following sibling -1.
Set the seq of the moving item to the current timestamp.

After moving item 105 to preceed item 102 the list would look like:

(1, 1, 101)
(1, 6, 105) # items with duplicate weight sorted on seq
(2, 2, 102)
(3, 3, 103)
(4, 4, 104)

We were able to move item 105 without having to change any other items in the list. And it almost looks like we did it just by chaning one value. However we cannot change column names in Cassandra, we can only create and delete them. This read-modify-write pattern has serious implications when it comes to concurrent changes as we’ll see later. Until then let’s implement the current design.

To store natually sorted lists in Cassandra start with a cassandra-cli schema:

To load and modify the list using Python and pycassa:

Note that the move() function deletes the old column before inserting the new one.

To exercise the code I first called initialise() to fill the list and used get() to view it. I then called move() to place item 5 before 2. When item 4 is moved to be before 2 it is given then same weight as 5 but a higher seq. This places 4 after 5 and both of them before 2:

We moved the item but it was not pretty; we moved it by deleting and then re-inserting. If our read-modify-write operation for item 5 overlaped with another client moving 5 it would result in two entries for 5. In a tradtional RDBMS we could prevent this by running an ACID Transaction working at Repeatable Read or better. But this is Sparta / Cassandra (cross out as applicable) and we don’t have locking transactions.

The timeless wonder of the whole thing

The overlapping operations we are trying to avoid look like this:

If the Clients used a REPEATABLE READ Transaction around all their work they would block each other. The Shared locks taken when reading would prevent the other Client from obtaining the Exclusive lock needed to delete. The DB Engine would pick an victim and ABORT it’s Transaction.

A Transaction around the Delete and Write calls would work. The Delete from Client 1 would be blocked until Client 2 completed. Client 1 could then notice it’s Delete updated 0 rows and ABORT it’s own Transaction.

The locks taken in the Transaction would prevent the Clients from modifying the same data. This is known as serializability and is the purpose of Transactions:

…a transaction schedule is serializable if its outcome (e.g., the resulting database state) is equal to the outcome of its transactions executed serially, i.e., sequentially without overlapping in time.

I can think of two approaches to making our move() operations super cereal. The first is to make the Cassandra requests inherently serialisable. The second is to record the move() transformations and have the application apply them in a serial fashion.

Take It and Turn It

At the start of the post I restricted the problem to lists of 50 “things”. I chose 50 for no particular reason. So instead of 50 let’s say lists where it’s reasonable to read all of the items at once. Even if you don’t want to display all of the items. To get idea of what is reasonable, I can read 50 Columns that are 50 bytes long from a hot row cache in about 260 Microseconds (excluding network IO).

While we are at it lets say it’s unnecessary to have Cassandra maintain the order of the items in the list. It won’t take much effort for the client to sort them. So lets see what happens when we give up the ability to ask the database for exactly the right data in the correct order.

The weight, seq and label for items in the Read Sorted list will be stored in Column values. While the Column names will contain the item_id and a property name:

To move an item we update it’s weight and seq Columns, and because they are column values we do not need to delete first. Row Level Isolation ensures that reading clients see either all updates or none. Overlapping requests from clients that update the same row will be serialised by the server. And the Column Timestamp included in the request means that the order they are applied in does not matter. Moving an item is now an atomic and serialisable operation.

Read Sorted lists are implemented by the Python code:

Note that move() now makes a single insert call.

I exercised the code using the same steps as above:

The output from the cassandra-cli shows that the order of the Columns does not change when items are moved.

Ledger List

It’s time to start adding functionality, and complexity. The Ledger List maintains the order of items in Cassandra Columns, so uses a delete and insert for moving. The move operations are written to an application Transaction Log rather then applied directly to the list. When a client reads the list it applies the current transactions to it’s client local copy; a process that is inherently serializable. Updates to the list stored in Cassandra are done through background worker processes which can be syncronised at the application level. The Data Model is roughly the same as Natural Lists so lets start there.

Each item will usually be represented by one column in a row in the LedgerList Column Family. The column name will be a composite of weight, seq, item_id and deleted. The column value will be used to store the item label. The deleted component is used to soft delete an item in the list when applying a Transaction. When an item is soft deleted a new column is created with the same (weight, seq, item_id) that has deleted set to True. Transactions are serialised as JSON and stored in a single column where the column name is the Transaction ID:

The move() function creates a Transaction and stores it in the LedgerTransactions Column Family. When get() reads the list it applies the Transactions to it’s local copy. Later a background process can call apply_tx() to update the list in Cassandra. While both get() and apply_tx() move items their implementation is very different.

apply_tx() has a tough life; to move an item it must delete the old column and insert a new one. It needs to make sure that Clients see either the state before the move or after it. But the Row Level Isolation guarantee used previously cannot be relied on, as it only applies to a single row mutation. And deleting and inserting columns for the same row must be done with two mutations. So our approach cannot rely on a delete and an insert being processed together.

To apply a Transaction to the list in Cassandra we:

Soft delete the old item and insert the new one by inserting new columns.
Hard delete the old item and the soft delete marker.
Hard delete the Transaction.

Once the insert in step 1 has completed was can consider the Transaction “in flight” until step 3 completes. During this time there may be three columns in the row that identify the item we are moving. The original entry will still be there with it’s original weight and seq. A soft delete column will be there with the same weight and seq as the original, but marked as deleted. And finally a second non deleted column will be there placing the item at it’s new position. After step 3 the Transaction has been deleted and there will only be a single column for the item again. Reads that take place while a Transaction is in flight have to account for the intermediate state they are seeing.

When get() reads the list it has to apply all the Transactions to it’s local copy, excluding those that are in flight. For each Transaction record it:

Checks if the item being moved has been soft deleted. If so the Transaction is in flight (after step 1 in apply_tx()) and can be skipped.
Tries to remove the item being moved from it’s local copy. If the item is missing the Transaction is still in flight (between steps 2 and 3 in apply_tx()) and can be skipped.
Inserts the item at the new location.
Filters to remove soft deletes left by step 1.
Sorts the list.

Ledger Lists are implemented by the Python code below which includes some diagnostic print statements:

To exercise the code I used the ISO list of countries found in the gist that contains the sample code. First the list was initialised and a couple of countries moved:

Note that when an item is moved the row in the LedgerList Column Family does not change.

Next I ran apply_tx() and passed a flag to stop processing after step 1. This places the Transaction in flight by inserting the soft delete for the old item and writing the new one. Log messages from get() show it detected the in flight Transaction and skipped processing:

A second call to apply_tx() had it re-process the in flight Transactions and stop at step 2. Log messages from get() show it detected the items had already been moved and skipped processing:

A final call to apply_tx() without any parameters processed the Transactions to completion. get() no longer has to apply Transactions:

What’s Missing ?

The next things to look at are pagination and adding more operations such as “add” and “delete”. It would also be handy to move items outside of the current page.

So far this is has been an experiment conducted on my couch, for fun. If you think I have missed something or got it wrong let me know.

Row Level Isolation and Consensus

2012-08-16T00:00:00-07:00

Row Level Isolation in Cassandra 1.1 is the most important new feature in Cassandra so far. There have been a lot of great improvements since version 0.5, but row level isolation adds an entirely new feature. One that opens the door to new use cases.

But if you could just see the beauty

If you are not familiar with row level isolation Data Stax have a blog post on it. In addition I covered some of the implications in my Cassandra Summit 2012 talk on Query Performance.

To recap, row isolation leverage’s the “zero copy clone” feature of Snap Tree. This allows write threads in Cassandra to clone the current Memtable columns for a row and work on the cloned copy. Changes are then only applied to the Memtable if the row has not been updated while the write thread was working.

In my Cassandra Summit talk I replicated the logic as best I could using Commodore 64 basic:

00 REM Cassandra for C64
05 REM Clone row_cols into my_cols
10 GOSUB 1000
20 FOR col = write.first_col TO write.last_col
25 REM Add or Reconcile col with my_cols
30 GOSUB 2000
40 IF my_cols != row_cols THEN GOTO 05
50 NEXT col
55 REM Atomic swap row_cols with my_cols
60 GOSUB 3000
70 IF swapped_cols = FALSE THEN GOTO 05

At line 10 we clone the current Memtable columns for the row we want to update. Each column in the mutation is then reconciled against the my_cols copy, isolating the changes from the Memtable. After each column is added the code checks if the Memtable’s row_cols object has been replaced (line 40). If it has we jump back to 10 and try our changes again. A final check in the form of an atomic compareAndSet is done at line 60. Again if the Memtable row_cols has been replaced this write thread tries again.

As a result readers will see either all or none of the changes applied by a writer. It also means that writer threads may take several attempts to complete their write. In my talk I looked at clients trying to write 10,000 (50 byte) Columns to a row in chunks of 50 columns. For the grey line below 10 clients wrote to different rows, for the red line they wrote to the same row.

There are some trade off’s. The increased latency is due to write threads detecting changes to the Memtable and restarting their operation. On the upside however we now have a database that provides lock free atomic assignment to multiple registers / values.

(AFAIK we cannot call this wait free as there is no guarantee that a particular write thread will complete given infinite number of steps. For example a client writing 1,000 columns may be continually stopped by clients writing 1 column.)

We Can All Agree on Consensus

Consensus is an interesting topic in distributed systems, and one I’m enjoying learning about from The Art of Multiprocessor Programming. It’s an excellent book, and the inspiration for this experiment. From page 100:

A consensus object provides a single method decide(). Each thread calls the decide() method with its input v at most once. The object’s decide() method will return a value meeting the following conditions:

consistent: all threads decide the same value.

valid: the common decision value is some threads input.

A further restricted placed on the algorithm is that is wait free, and there for lock free.

A Consensus Protocol has a Consensus Number n that describes the maximum number of threads it provides consensus for. Obviously the simplest level of Consensus is between two threads, so lets look at a protocol with a Consensus Number of 2.

Assign23

2-Consensus with a “(2/3) array” is discussed in the book which credits the idea to Maurice Herlihy, Wait-Free Synchronisation, 1991 (in case you missed it Maurice Herlihy is one of the book authors). It is also discussed in the teaching slide deck for chapter 5. The (2/3) array is actually used to prove it’s impossible to implement (wait free, so lock free) 2-consensus using only atomic registers.

The Assign23 class from the book looks like this:

Assign23 wraps access to an array of three elements. It uses a lock via the synchronized keyword to allow a caller to atomically write 2 array elements at a time. It uses the same lock when providing read access to ensure reads are consistent. That is reads only see the state before two elements are updated or after. The effect is to isolate writes from reads.

To build a 2-consensus protocol using Assign23 Threads write to elements in the array and then observe the state of the array. Each Thread (remember we only have 2) writes to 2 elements in the array. Thread 0 writes 1 to elements 0 and 1, while Thread 1 writes 1 to elements 1 and 2. To decide the outcome both Threads read part of the array. The consensus we are after is which Thread wrote to the array first.

Lets say we initialise an instance of Assign23 with -1 in all array elements. If Thread 0 writes and then reads from the array it will see three possible states:

From Thread 0s point of view it will decide:

Case 1: Thread 0 wins. Thread 1 has not not written to the array.
Case 2: Thread 0 wins. Thread 0 wrote first, then Thread 1.
Case 3: Thread 1 wins. Thread 1 wrote first, then Thread 0.

The 2-consensus protocol is then constructed as:

Note that decide() does not tell you which thread wrote it’s value to the proposed array first. Or which made the call to assign() first. The Thread that is considered the winner is the one that acquired the lock (on the instance) implemented by the synchronized keyword.

CF23

So to implement Assign23 all we needed was atomic, isolated, read-write access to an array. Hmmm, I think we can do that in Cassandra.

Start with a Column Family definition using CQL3:

(Defining the column names up front is not necessary. It just makes the select statements below look nicer.)

The row key for the cf23 Column Family is the name of the Consensus object. The element_* columns represent the r array from Assign23, and the propose_* columns are the values proposed by the threads / clients.

To make the example easier let’s use Python and pycassa.

To use the Consensus2 class two clients needs to know the name of the consensus object they are sharing, and their client id. With that they can propose a value and agree on which value “won”.

In the examples below I ran two iPython clients and examined the database using cqlsh. In the first test player 1 answers the question first:

In the second test player 2 gets in first:

The Fine Print

Is this fair? Does the client who calls Cassandra first always win? No.

For example say the write thread for client 0 starts executing nanoseconds before the write for client 1. The first write thread may be pre-empted by the JVM before it gets a chance to complete, allowing the second thread to finish first. When the JVM resumes the first writer it will discover that the row columns have changed and re-apply it’s mutation. Unfortunately by then it’s too late, client 1 has won. A small “wall clock” advantage in starting a write is not enough to win. The race between the clients is the first to pass the GOSUB call at line 60 of our C64 code above.

This being Cassandra there is still a potential issue with timestamps and clock skew between clients. The value of element_1 in the cf23 row depends on the timestamps used by clients. If client 0 writes with a timestamp skewed to the future it’s write for element_1 may always win when reconciled with client 1. If this happened when client 0 wrote first, the array would look like Case 3 (from above) rather than Case 2. The result would be client 1 “winning” when really client 0 won.

In this situation CQL has an advantage over the RPC interface. If a TIMESTAMP is not included in an INSERT statement the coordinating node generates a timestamp for the new Columns. Baring NTP moving the clock backwards, successive calls to the same server should result in monotonic timestamps.

Finally you cannot reuse the Consensus2 object, the same is true for the in memory Java version in the book. If you do weird things happen:

ZooKeeper Reading 12-01-2012

2012-01-12T00:00:00-08:00

Overview

There is a leader in the cluster.
Hierarchical model of znodes which act like directories and files. Uses ”/” as the path separator.
All in memory, high throughput low latency.
“The ZooKeeper implementation puts a premium on high performance, highly available, strictly ordered access.”
Transaction logs and snapshots on disk.
Clients hold a TCP connection for duplex messaging.
All updates have a globally ordered TxID.
Works best in read heavy workloads, think 10:1 R:W ratios.
znodes may have data and children.
znodes have a version number for their local (?) state.
Reads and writes on a znode are atomic with respect to the version of data.
znodes can be protected by an ACL.
Ephemeral znodes are deleted when the session that created them ends.
Clients can set a watch on znode that is triggered when it changes. Is this for the local data or local data and children ?
Guarantees: * Sequential Consistency - Updates from a client will be applied in the order that they were sent. * Atomicity - Updates either succeed or fail. No partial results. * Single System Image - A client will see the same view of the service regardless of the server that it connects to. * Reliability - Once an update has been applied, it will persist from that time forward until a client overwrites the update. * Timeliness - The clients view of the system is guaranteed to be up-to-date within a certain time bound.
Write to WAL before apply to in memory DB.
Reads are serviced by the local DB on the node, writes are services by an agreement protocol. Guess this is why it’s tuned for read.
Writes go to the leader, are then distributed to the other (follower) nodes. Local DB’s should never diverge.
Performance 3 servers should give between 20k/sec and 80k/sec requests depending on the read/write mix.
Reliability less than 200ms to elect a new leader, failure of a follower reduces throughput.
What’s the recover model for a follower that is down for a while ? Does this affect performance ? Answer (from the internals) If too many Proposals are missing a snapshot is sent.

ZooKeeper Internals

“At the heart of ZooKeeper is an atomic messaging system that keeps all of the servers in sync.”
Guarantees: * Reliable Delivery * Total Order * Causal Order
The messaging layer is build around FIFO channels between nodes, and relies on the properties of TCP for this. Specifically: * Ordered delivery * No message after close
The protocol is composed of: * Packet: a sequence of bytes sent through a FIFO channel * Proposal: a unit of agreement. Proposals are agreed upon by exchanging packets with a quorum of ZooKeeper servers. Most proposals contain messages, however the NEW_LEADER proposal is an example of a proposal that does not correspond to a message. * Message: a sequence of bytes to be atomically broadcast to all ZooKeeper servers. A message put into a proposal and agreed upon before it is delivered.
QUORUM is (n/2) +1 by default.
QUORUM can be majority quorums, weights, or a hierarchy of groups.
Proposals are stamped with the zxid and sent to all servers, a server ack’s when it is on persistent store. Messages in the proposal are then delivered.
zxid has two parts: the epoch and a counter. Implemented as a 64 bit int, high 32 bits are the epoch, low 32 are the count.
“The epoch number represents a change in leadership. Each time a new leader comes into power it will have its own epoch number.”
Messaging consists of two phases, Leader Activation and Active Messaging.
Leader Activation may appear to have worked but later fail when checking the invariant that a QUORUM of followers follow the same leader. During the election it must only hold with a high probability.
In Active Messaging: * Leader sends PROPOSE to all followers for a new proposal. * Followers commit to non-volatile storage and then ACK * Leader sends COMMIT to all followers once a QUOURM have ACK‘d.

Getting Started

Grab the latest distro and start a single node with bin/zkServer.sh start-foreground conf/zoo_sample.cfg
Fire up the command line interface with bin/zkCli.sh -server 127.0.0.1:2181 and work through the examples in the doc.

Zope ZooKeeper client for Python

Requires zc-zookeeper-static
zc-zookeeper-static is a wrapper around the C libs, it’s pretty low level. e.g. you get an int handle and pass that into methods, not OO. zc.zk ads an OO wrapper and some other stuff I cannot work out.
A lot of methods on the zc.zk.ZooKepper object are pass through to the zc-zookeeper-static package and do not have any docs. Check to docs on zookeeper for the function help. For example zc.zk.ZooKeeper.get has no docs and a crap (*arg, **kwargs) param list, look at zookeeper.get.

Get a connection

import zc.zk
zk = zc.zk.ZooKeeper('127.0.0.1:2181')

Get the children of a `znode`

In [6]: zk.get_children("/")
Out[6]: ['consumers', 'brokers', 'zookeeper', 'zk_test']
# some stuff from kafka there.

Get the properties of a `znode`

# Get a zc.zk.Properties for the path
# **NOTE:** This is heavy weight, for single reads use get_properties()
In [55]: p = zk.properties("/zookeeper")

In [56]: p.data
Out[56]: {}

In [58]: p.values()
Out[58]: []

# simple get_properties()
In [64]: zk.get_properties("/zk_test")
Out[64]: {'string_value': 'foo'}

#zv.zk assume node data is json
In [49]: zk.set("/zk_test", "foo")
Out[49]: 0

In [51]: p = zk.properties("/zk_test")

In [53]: p.values()
Out[53]: ['foo']

In [54]: p.data
Out[54]: {'string_value': 'foo'}

Tree operations

In [59]: zk.print_tree("/zookeeper")
/zookeeper
  /quota

`znode` operations

# create an ephemeral node

# must have an ACL this is an open one 
In [78]: acl = [{"perms" : zookeeper.PERM_ALL, "scheme" : "world", "id" : "anyone"}]

# Parent path must exist
In [85]: zk.create( "/fake/ephemeral", "some data", acl, zookeeper.EPHEMERAL)
...
NoNodeException: no node

In [84]: zk.create( "/zk_test/ephemeral", "some data", acl, zookeeper.EPHEMERAL)
Out[84]: '/zk_test/ephemeral'

# node now listed (locally) on the connection 
In [86]: zk.ephemeral
Out[86]: 
{'/zk_test/ephemeral': {'acl': [{'id': 'anyone',
                                 'perms': 31,
                                 'scheme': 'world'}],
                        'data': 'some data',
                        'flags': 1}}
                        
# View from the cluster
In [88]: zk.get_properties("/zk_test/ephemeral")
Out[88]: {'string_value': 'some data'}

In [89]: p = zk.properties("/zk_test/ephemeral")
In [91]: p.meta_data
Out[91]: 
{'aversion': 0,
 'ctime': 1326337991257L,
 'cversion': 0,
 'czxid': 1950L,
 'dataLength': 9,
 'ephemeralOwner': 86922380708675587L,
 'mtime': 1326337991257L,
 'mzxid': 1950L,
 'numChildren': 0,
 'pzxid': 1950L,
 'version': 0}

Watch for changes

In [8]: children = zk.children("/zk_test")

In [9]: def my_callback(node):
   ...:   print "Called with node: ", str(node)
   ...: 

In [11]: children(my_callback)
Called with node:  zc.zk.Children(0, /zk_test)
Out[11]: zc.zk.Children(0, /zk_test)

In [14]: acl = [{"perms" : zookeeper.PERM_ALL, "scheme" : "world", "id" : "anyone"}]

In [15]: zk.create( "/zk_test/ephemeral", "some data", acl, zookeeper.EPHEMERAL)
Out[15]: '/zk_test/ephemeral'
Called with node:  zc.zk.Children(0, /zk_test)

Cassandra Reading 02-01-2012

2012-01-02T00:00:00-08:00

CQL: SQL In Cassandra

And a blog post

By Eric Evans

Good examples of simple statements.
Performance comparisons show CQL has 5% to 10% lower throughput and higher latency.
List of drivers and where to get them.
More improvements coming.

Storm Cassandra Integration

By P. Taylor Goetz

A storm bolt to persist data to Cassandra.

Cassandra in Online Advertising: Real Time Bidding

By Edward Capriolo

10TB of data and growing.
Latency requirements of less than 120ms.
“Distributed not Duplicated” is a great phrase.
Ed still loves the Cacti :)
Good idea to use JMX to modify the (Cassandra) cache settings for a single node to compare against the others.
Pay attention to how the cache hit rate varies with regard to the cache size, at some point may be better to accept a (say) 90% hit rate and give more memory to another CF.
Tuning IO performance for peak and off-peak by modifying nodetool setcompactionlimit to improve compaction performance.
Running night time major compactions like Urban Airship. I wonder how leveled compaction would work with the mixed workload?

Replication and the latency-consistency tradeoff

By Daniel Abadi

A discussion about latency and consistency.
“there’s no way to perform consistent replication across database replicas without some level of synchronous network communication.”
Not sure I agree that in Dynamo / Cassandra (not sure about Riak) “updates generally go to the same node, and are then propagated synchronously to W other nodes (case (2)(c))”.
I think how inconsistencies are handled during read requests is another source of latency.

Cassandra for sys admins

By Nathan Milford

14 nodes in two DC’s
In productions since Cassandra version 0.4, awesome!
A good list of things to monitor in a cluster.
Good best practices for shutting down a node that give the fastest startup, also see.

Cassandra for LOBS

By Dan Pritchett

Expensive SAN storage bombshell.

Expedia Hotel Price Cache

By B. Todd Burruss

Pre-calculate, trading space for time.
A rolling window of 2.8 billion data points.
Test, measure, tune.

Data Modeling Examples

By Matthew Dennis

I always check out Matthew’s data model presentations to see what the best practices are.
“Usually better to keep a record that something happened as opposed to changing a value”.
Good advice on time series and the XACT_LOG.

Cassandra In Production: Things We Learned

By Walt Jones

Using Cassandra 0.7, there are a lot of improvements in Cassandra 1.0.
12 AWS EC2 m1.xlarge nodes with 5TB of data.
S3 archive
The memory footprint issues has been eliminated in Cassandra 1.0.
Please avoid using Super Columns.
Please use the Random Partitioner.

Data Modeling with Cassandra

By Sam Overton

De-normalize for a brighter future.
No SQL “Hello World” twitter example.

Anatomy of a Cassandra Partition

2011-12-15T00:00:00-08:00

Recently I was working on a Cassandra cluster and experienced a strange situation that resulted in a partition of sorts between the nodes. Whether you actually call it a partition or not is a matter for discussion (see ” You Can’t Sacrifice Partition Tolerance - Updated October 22, 2010”]. But weird stuff happened, Cassandra remained available, and it was fixed with zero site down time. It was also a good example of how and why Cassandra is a Highly Available data store, and fun to fix. So here are all the nerdy details…

The cluster was running Cassandra 1.0.5 on 6 nodes, 3 in the East DC (nodes 20, 21 and 22) and 3 in the West DC (nodes 23, 24 and 25), and the Keyspace was configured with RF 3 in each DC. Application servers were running in each DC and it was important to keep a LOCAL_QUOURM in each data centre. The code would drop back to QUORUM if necessary, but that was something we wanted to avoid.

I was deploying some configuration changes that required a rolling restart. The first few nodes worked fine, after restarting Cassandra I watched the other nodes detect it was UP and start sending hints to it. However when node #23 restarted the rest of the cluster continued to see it as DOWN, even though node #23 could see the rest of the cluster as UP.

Is Cassandra available ?

This was the first question to answer. For this situation available meant all keys could be written to or read from using the LOCAL_QUOURM CL. Technically as I still had at least 2 nodes in each data centre I still had a Quorum, but strange stuff was happening so I wanted to confirm this.

Node #23 appeared to be happy:

cassandra23# bin/nodetool -h localhost info
Token            : 28356863910078205288614550619314017621
Gossip active    : true
Load             : 275.44 GB
Generation No    : 1762556151
Uptime (seconds) : 67548
Heap Memory (MB) : 2926.44 / 8032.00
Data Center      : DC1
Rack             : RAC_unknown  
Exceptions       : 0

(Note: this is from a little later, so the Uptime is not correct)

When I looked at the ring from any node other than #23 all nodes were UP other than node #23 (10.29.60.10):

cassandra20# nodetool -h localhost ring
Address         DC          Rack        Status State   Load            Owns    Token                                       
                                                                               141784319550391026443072753096570088106     
10.37.114.8     DC1         RAC20       Up     Normal  285.86 GB       16.67%  0                                           
10.29.60.10     DC2         RAC23       Down   Normal  277.86 GB       16.67%  28356863910078205288614550619314017621      
10.6.130.70     DC1         RAC21       Up     Normal  244.9 GB        16.67%  56713727820156410577229101238628035242      
10.29.60.14     DC2         RAC24       Up     Normal  296.85 GB       16.67%  85070591730234615865843651857942052864      
10.37.114.10    DC1         RAC22       Up     Normal  255.81 GB       16.67%  113427455640312821154458202477256070485     
10.29.60.12     DC2         RAC25       Up     Normal  316.88 GB       16.67%  141784319550391026443072753096570088106

But when I looked at the ring from node #23 all nodes were UP:

cassandra23# nodetool -h localhost ring
    Address         DC          Rack        Status State   Load            Owns    Token                                       
                                                                                   141784319550391026443072753096570088106     
    10.37.114.8     DC1         RAC20       Up     Normal  285.86 GB       16.67%  0                                           
    10.29.60.10     DC2         RAC23       Up     Normal  277.86 GB       16.67%  28356863910078205288614550619314017621      
    10.6.130.70     DC1         RAC21       Up     Normal  244.9 GB        16.67%  56713727820156410577229101238628035242      
    10.29.60.14     DC2         RAC24       Up     Normal  296.85 GB       16.67%  85070591730234615865843651857942052864      
    10.37.114.10    DC1         RAC22       Up     Normal  255.81 GB       16.67%  113427455640312821154458202477256070485     
    10.29.60.12     DC2         RAC25       Up     Normal  316.88 GB       16.67%  141784319550391026443072753096570088106

So it did not appear to be a cross DC networking issue, and technically we have a Quorum of replicas in each data centre.

To confirm the consistency levels available to the application I opened the cassandra-cli on nodes #20 and #23 and did some get’s changing the consistency level with consistencylevel as. It was still working but things were strange, this is what I found.

Node #23 could serve requests at LOCAL_QUORUM, QUORUM and ALL consistency.
All other nodes could serve requests at LOCAL_QUOURM and QUORUM but not ALL consistency.

Also I could see requests been processed by node #23 using nodetool tpstats.

So from the point of view of the application the cluster was up, and the application guys confirmed this. However we had lost redundancy in the West DC, if we lost another node for any reason we would lose the LOCAL_QUORUM. Any reason can include things like a planned restart.

To be sure this had to be fixed, but we had time to fix it. This is part of the ops story with Cassandra, if something goes wrong you can normally fix it while the application keeps working.

Orientate

My initial thought was that either the Gossip heartbeat from node #23 was not getting through, or it was been ignored. So I started looking around to get an idea of what view the nodes in the cluster had of node #23.

Nodes in the rest of the cluster could telnet to node #23 on port 7000 the default storage_port. A simple test but it helps to know the basic assumptions are still valid.

All the nodes (including #23) had a consistent view of the state of the other nodes propagated via Gossip, for example:

cassandra20# bin/nodetool -h localhost gossipinfo
...
/10.29.60.10
  LOAD:2.98347080902E11
  STATUS:NORMAL,28356863910078205288614550619314017621
  RPC_ADDRESS:10.29.60.10
  SCHEMA:fe933880-19bd-11e1-0000-5ff37d368cb6
  RELEASE_VERSION:1.0.5

The “Application State” transmitted via Gossip is not directly related to a node been marked as DOWN (more later), but at least I could see that Gossip was distributing the same information about the node around the cluster. The properties listed above have the following meaning:

LOAD: The total live disk space used, in bytes.
STATUS: The ring state of the node, it’s token (if any) and some other information. NORMAL indicates the node is a normal member of the ring, the other options are BOOT, LEAVING, LEFT, MOVING.
RPC_ADDRESS: IP address to connect to this node on.
SCHEMA: UUID of the schema this node is using.
RELEASE_VERSION: Cassandra version.

If Application State was responsible nodetool ring would have said something other than DOWN for node #23.

Node #23 should have been trying to tell the other nodes that it is here and happy to answer queries. So the next step was to look at the Gossip traffic and see if that was happening. I started by enabling TRACE debugging on the main org.apache.cassandra.gms.Gossiper class by adding the line below to conf/log4j-server.properties, log4j is configured to watch for changes so you can just change the file and wait a few seconds.

log4j.logger.org.apache.cassandra.gms.Gossiper=TRACE

To disable a live change to the logging config like this it’s not enough to just comment out the line. You will need to a provide a new value for the Gossiper logger. So after I saw what I wanted in /var/log/cassandra/system.log on node #20 I changed the line the logging config to be:

log4j.logger.org.apache.cassandra.gms.Gossiper=INFO

This was the first thing I looked at in the logs on node #20:

TRACE [GossipStage:1] 2011-12-13 00:58:49,636 Gossiper.java (line 647) local heartbeat version 526912 greater than 7951 for /10.29.60.10
TRACE [GossipStage:1] 2011-12-13 00:58:49,636 Gossiper.java (line 661) Adding state LOAD: 2.98347080902E11
TRACE [GossipStage:1] 2011-12-13 00:58:49,636 Gossiper.java (line 647) local heartbeat version 1231 greater than 1229 for /10.37.114.10
TRACE [GossipStage:1] 2011-12-13 00:58:49,636 Gossiper.java (line 647) local heartbeat version 4892 greater than 4890 for /10.29.60.12

The first line was the important one, something in the cluster was telling node #20 about node #23. The version number referred to in the message is part of the heartbeat, a higher version number means the heartbeat is more recent. So it looked like node #20 was remembering something about node #23 from one of the machines that had been restarted, that would explain the higher local version number. But that didn’t make sense, Cassandra can handle nodes restarting.

This was not the right place to be looking. The log message could have been logged in response to information about node #23 that was sent from any other node. I needed to find proof of Gossip messages been received directly from node #23.

Gossip uses a 3 step protocol, the initiating node sends an Syn message, the target replies with an Ack and the initiator replies with an Ack2. So I went hunting information about Syn’s received from node #23 by enabling TRACE logging on the verb handler:

log4j.logger.org.apache.cassandra.gms.GossipDigestSynVerbHandler=TRACE

This is what I found in the node #20 logs:

TRACE [GossipStage:1] 2011-12-13 02:04:56,165 GossipDigestSynVerbHandler.java (line 46) Received a GossipDigestSynMessage from /10.29.60.10
TRACE [GossipStage:1] 2011-12-13 02:04:56,166 GossipDigestSynVerbHandler.java (line 76) Gossip syn digests are : /10.6.130.70:1323732220:9792 /10.37.114.10:1323736718:5242 /10.29.60.12:1323733099:8904 /10.29.60.10:1762556151:11964 /10.29.60.14:1323732392:9619 /10.37.114.8:1323731527:10494 
TRACE [GossipStage:1] 2011-12-13 02:04:56,166 GossipDigestSynVerbHandler.java (line 90) Sending a GossipDigestAckMessage to /10.29.60.10

It got a Syn from node #23 and replied with an Ack. It was receiving and responding to messages from node #23 but it still thought it was DOWN. Things were getting weirder, time to go check the code.

Right after logging that message GossipDigestSynVerbHandler notifies the FailureDetector that it had heard from the endpoint. The FailureDetector is ultimately responsible for deciding if a node is UP or DOWN, the clue is in the name. GossipDigestSynVerbHandler actually calls the main Gossiper class which does a few things and then decides if it wants to report to the FailureDetector. I just wanted to know if the call was made, so I enabled the logging again:

log4j.logger.org.apache.cassandra.gms.GossipDigestSynVerbHandler=TRACE
log4j.logger.org.apache.cassandra.gms.FailureDetector=TRACE

And found this in the logs on node #20:

TRACE [GossipStage:1] 2011-12-13 02:14:37,033 GossipDigestSynVerbHandler.java (line 46) Received a GossipDigestSynMessage from /10.29.60.10
TRACE [GossipStage:1] 2011-12-13 02:14:37,033 GossipDigestSynVerbHandler.java (line 76) Gossip syn digests are : /10.29.60.10:1762556151:12552 /10.29.60.14:1323732392:10208 /10.37.114.8:1323731527:11082 /10.37.114.10:1323736718:5830 /10.6.130.70:1323732220:10379 /10.29.60.12:1323733099:9493 
TRACE [GossipStage:1] 2011-12-13 02:14:37,033 GossipDigestSynVerbHandler.java (line 90) Sending a GossipDigestAckMessage to /10.29.60.10
TRACE [GossipStage:1] 2011-12-13 02:14:37,037 GossipDigestSynVerbHandler.java (line 46) Received a GossipDigestSynMessage from /10.37.114.10
TRACE [GossipStage:1] 2011-12-13 02:14:37,037 GossipDigestSynVerbHandler.java (line 76) Gossip syn digests are : /10.37.114.10:1323736718:5831 /10.29.60.12:1323733099:9493 /10.6.130.70:1323732220:10379 /10.29.60.14:1323732392:10208 /10.37.114.8:1323731527:11082 /10.29.60.10:1762556151:526912 
TRACE [GossipStage:1] 2011-12-13 02:14:37,037 GossipDigestSynVerbHandler.java (line 90) Sending a GossipDigestAckMessage to /10.37.114.10
TRACE [GossipStage:1] 2011-12-13 02:14:37,038 FailureDetector.java (line 164) reporting /10.37.114.10
TRACE [GossipStage:1] 2011-12-13 02:14:37,038 FailureDetector.java (line 164) reporting /10.29.60.12
TRACE [GossipTasks:1] 2011-12-13 02:14:37,447 FailureDetector.java (line 185) PHI for /10.29.60.12 : 0.1740028340787402
TRACE [GossipTasks:1] 2011-12-13 02:14:37,447 FailureDetector.java (line 185) PHI for /10.6.130.70 : 0.17769446955863263
TRACE [GossipTasks:1] 2011-12-13 02:14:37,447 FailureDetector.java (line 185) PHI for /10.29.60.14 : 0.3766607052738764
TRACE [GossipTasks:1] 2011-12-13 02:14:37,447 FailureDetector.java (line 185) PHI for /10.37.114.10 : 0.16292431576785454
TRACE [GossipTasks:1] 2011-12-13 02:14:37,448 FailureDetector.java (line 185) PHI for /10.29.60.10 : 9511.26282656631
TRACE [GossipTasks:1] 2011-12-13 02:14:37,448 FailureDetector.java (line 189) notifying listeners that /10.29.60.10 is down
TRACE [GossipTasks:1] 2011-12-13 02:14:37,448 FailureDetector.java (line 190) intervals: 500.0 mean: 500.0
TRACE [GossipTasks:1] 2011-12-13 02:14:38,453 FailureDetector.java (line 185) PHI for /10.29.60.12 : 0.6019902450401402
TRACE [GossipTasks:1] 2011-12-13 02:14:38,453 FailureDetector.java (line 185) PHI for /10.6.130.70 : 0.5983077316197724
TRACE [GossipTasks:1] 2011-12-13 02:14:38,453 FailureDetector.java (line 185) PHI for /10.29.60.14 : 0.7697319392007641
TRACE [GossipTasks:1] 2011-12-13 02:14:38,453 FailureDetector.java (line 185) PHI for /10.37.114.10 : 0.5636623638423329
TRACE [GossipTasks:1] 2011-12-13 02:14:38,454 FailureDetector.java (line 185) PHI for /10.29.60.10 : 9512.137495652863
TRACE [GossipTasks:1] 2011-12-13 02:14:38,454 FailureDetector.java (line 189) notifying listeners that /10.29.60.10 is down
TRACE [GossipTasks:1] 2011-12-13 02:14:38,454 FailureDetector.java (line 190) intervals: 500.0 mean: 500.0
TRACE [GossipStage:1] 2011-12-13 02:14:38,454 FailureDetector.java (line 164) reporting /10.29.60.14
TRACE [GossipStage:1] 2011-12-13 02:14:38,454 FailureDetector.java (line 164) reporting /10.37.114.10
TRACE [GossipStage:1] 2011-12-13 02:14:38,454 FailureDetector.java (line 164) reporting /10.6.130.70
TRACE [GossipStage:1] 2011-12-13 02:14:38,454 FailureDetector.java (line 164) reporting /10.29.60.12
TRACE [GossipStage:1] 2011-12-13 02:14:39,030 GossipDigestSynVerbHandler.java (line 46) Received a GossipDigestSynMessage from /10.6.130.70

The first line shows a Syn message received from node #23, later a response Ack was sent and the next Syn handled without reporting to the FailureDetector. By contrast look at the next Syn message from 10.37.114.10, it results in the FailureDetector been told (via the report() function) that we’ve head from both 10.37.114.10 and 10.29.60.12.

You can also see that the FailureDetector has logged some information about node #23 (10.29.60.10). At the end of a Gossip round the node asks the FailureDetector to interpret the information it has about each other node. The FailureDetector tracks the intervals between when it is told about a node, these values are interpreted every second to calculate the PHI value used to determine the nodes liveness. By default a node with a PHI higher than 8 is considered DOWN, as you can see node #23 was way DOWN.

At the time I did not pay any attention to the other log messages from the FailureDetector, I just wanted to know why report() was not been called.

Stop and think

This is what I knew:

All nodes in the cluster other than node #23 had a consistent view that #23 was down.
New “Application State” about node #23 transmitted via Gossip was been ignored as the nodes thought it was from the past.
Direct Gossip messages from node #23 were not been reported to the FailureDetector.

So I put out a call for help on the #Cassandra IRC channel and started poking around Gossiper.notifyFailureDetector() to see why it was not reporting messages from node #23.

My assumption was that the high heartbeat version number node #20 had locally for node #23 was the reason for not reporting to the FailureDetector, and line 724 reinforce this view of the world. The FailureDetector was not called because the local version number was higher than the remote one. Done. I saw what I wanted to see, the version number is the problem.

But I still didn’t know why this was happening, it was unlikely to be a bug in the Gossip code. Rolling restarts are standard practice, we do them all the time. About this time I got some help from Brandon Williams (driftx) from Data Stax in the IRC room. Getting help is always good, and Gossip is Brandon’s thing.

Generational issues

Heartbeat versions are only part of how Gossip works out if new information it receives really is new, or is old information delivered out of order. Sitting above the version number is the Generation. When a major change occurs on a node, such as a restart or a changing tokens, the Generation number is increased

When a node detects a higher Generation for another endpoint in a Syn Gossip message it clears the the FailureDetector information for endpoint. More importantly it takes action to get the most recent information about the node. First the node includes a request in the Ack it returns to the gossiper for all the information it has about the endpoint where the generation has increased. Second when processing the digests in the returned Ack2 message the local endpoint state is replaced with the new state.

When node #23 restarted it should have picked up a new Generation number and the other nodes in the cluster should have noticed the change and updated their local state. The simple way to test if this was happening was do another restart on node #23 and see what happened, yay for redundant clusters with no single point of failure.

While that was going on Brandon noticed that the Generation number for node #23 was strange. The number is initialized to the current UTC time in seconds when the server is first started. After than it is increased by 1 for every major change, such as a restart. The number shown in the logs and other places for node #23 was 1762556151, which equates to Fri, 07 Nov 2025 22:55:51 GMT.

This server was from the future.

Still is should have incremented the Generation on restart and maintained the forward trajectory of the space time continuum. The restart completed. Node #23 came back on line with the same generation number 1762556151.

It was a server from the future that was stuck in present which was really it’s past!

Back to the future

The current Generation for a node is stored in the LocationInfo CF in the System KS. Every time the Generation is bumped it’s read from there, incremented and written back. So the next step was to see what was actually stored in the CF using the cassandra-cli:

[default@system] list LocationInfo;                      
Using default limit of 100
-------------------
RowKey: Ring
=> (column=00, value=0a257208, timestamp=1323730075935)
=> (column=2aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa, value=0a068246, timestamp=1323730075923)
=> (column=40000000000000000000000000000000, value=0a1d3c0e, timestamp=1323730075897)
=> (column=55555555555555555555555555555555, value=0a25720a, timestamp=1323730075907)
=> (column=6aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa, value=0a1d3c0c, timestamp=1323730075944)
-------------------
RowKey: L
=> (column=436c75737465724e616d65, value=737069, timestamp=1320437246450000)
=> (column=47656e65726174696f6e, value=690e78f6, timestamp=1762556150811000)
=> (column=50617274696f6e6572, value=6f72672e6170616368652e63617373616e6472612e6468742e52616e646f6d506172746974696f6e6572, timestamp=1320437246450000)
=> (column=546f6b656e, value=15555555555555555555555555555555, timestamp=1762556150877)
-------------------
RowKey: Cookies
=> (column=5072652d312e302068696e747320707572676564, value=6f68207965732c2074686579207765726520707572676564, timestamp=1320437246470)
-------------------
RowKey: Bootstrap
=> (column=42, value=01, timestamp=1762556150877)

4 Rows Returned.
Elapsed time: 37 msec(s).

There were columns in the “L” row (the local node state) and the “Cookies” row that were written with a timestamp from the future (the ones that start with 17…). Any attempt to overwrite these columns would fail unless it used a higher time stamp.

Here’s the “L” row with proper formatting for the column names:

[default@system] assume LocationInfo comparator as ascii; 
Assumption for column family 'LocationInfo' added successfully.
[default@system] get LocationInfo['L'];
=> (column=ClusterName, value=737069, timestamp=1320437246450000)
=> (column=Generation, value=690e78f6, timestamp=1762556150811000)
=> (column=Partioner, value=6f72672e6170616368652e63617373616e6472612e6468742e52616e646f6d506172746974696f6e6572, timestamp=1320437246450000)
=> (column=Token, value=15555555555555555555555555555555, timestamp=1762556150877)
Returned 4 results.
Elapsed time: 2 msec(s).

Generation is hex 690e78f6, which is 1762556150 binary. When node #23 started it read this value, added 1 to it to get 1762556151 and wrote it back. However the timestamp used to write it back was the current time, which was less than the one on disk so the write was ignored during reads. So the next time it started it got the same Generation number.

Back to the present

The fix was to somehow get the Generation number to be increased. The approach I took was to delete the Generation column with the high time stamp, and purge it’s existence out of the SSTables. It wasn’t enough to just delete the column, I needed to delete it with a higher time stamp and then see that Tombstone was purged so that future writes with current time stamps would work as expected.

This approach was possible because the GC_GRACE_SECONDS on the LocationInfo CF is 0. So tombstones are committed to disk, but will be purged the first time compaction processes them. If GC_GRACE_SECONDS was higher I would have to either wait for the time to pass or temporarily reduce it.

The other possible options were a full cluster stop and restart passing the Dcassandra.load_ring_state=false option. Or some form of monkeying with the token and/or IP address.

Here are the steps I took to get there, I’ve included a few additional ones to illustrate what was happening with the tombstone.

Step 1 - Snapshot

Always backup data before you start exploring it’s limits, think of it as safe word. If it all get’s a little to scary just say the safe word and we can stop.

cassandra23# bin/nodetool -h localhost snapshot system -t pre-generation-delete
Requested snapshot for: system 
Snapshot directory: pre-generation-delete

I snap shot a lot when doing stuff like this so that I can roll back to any point.

Step 1 - Delete the Generation column

To specify a timestamp for the delete I used the pycassaShell bundled with the pycassa client, this is an interactive Python session with some handy tools.

Remember I wanted to delete the Generation column:

[default@system] get LocationInfo['L'];
=> (column=ClusterName, value=737069, timestamp=1320437246450000)
=> (column=Generation, value=690e78f6, timestamp=1762556150811000)
=> (column=Partioner, value=6f72672e6170616368652e63617373616e6472612e6468742e52616e646f6d506172746974696f6e6572, timestamp=1320437246450000)
=> (column=Token, value=15555555555555555555555555555555, timestamp=1323805490467000)
Returned 4 results.
Elapsed time: 2 msec(s).

So I needed to specify a timestamp higher than 1762556150811000:

cassandra23# ./pycassaShell -H 10.29.60.10 -k system
[I]: IPython not found, falling back to default interpreter.
----------------------------------
Cassandra Interactive Python Shell
----------------------------------
Keyspace: system
Host: 10.29.60.10:9160

Available ColumnFamily instances:
 * HINTSCOLUMNFAMILY          ( HintsColumnFamily )
 * VERSIONS                   ( Versions )
 * INDEXINFO                  ( IndexInfo )
 * LOCATIONINFO               ( LocationInfo )
 * MIGRATIONS                 ( Migrations )
 * NODEIDINFO                 ( NodeIdInfo )
 * SCHEMA                     ( Schema )

Schema definition tools and cluster information are available through SYSTEM_MANAGER.
>>> LOCATIONINFO.remove("L", columns=["Generation"], timestamp=1762556150811001)

The Generation column was no longer returned in query results:

[default@system] get LocationInfo['L'];
=> (column=ClusterName, value=737069, timestamp=1320437246450000)
=> (column=Partioner, value=6f72672e6170616368652e63617373616e6472612e6468742e52616e646f6d506172746974696f6e6572, timestamp=1320437246450000)
=> (column=Token, value=15555555555555555555555555555555, timestamp=1323805490467000)
Returned 4 results.
Elapsed time: 2 msec(s).

Step 3 - Snapshot

Another snapshot for good luck. This one will also flush the LocationInfo CF so all of our changes are on disk.

cassandra23# bin/nodetool -h localhost snapshot system -t post-generation-delete
Requested snapshot for: system 
Snapshot directory: post-generation-delete

Step 4 - Check the delete

This was not necessary, I knew what was going to happen later but I like to confirm that my expectations are correct and it helps illustrate how tombstones work. What I expected to see was two SSTables for LocationInfo, the first would have all the columns we saw originally and the second would have the tombstone for the Generation column delete.

First check which SSTables are on disk:

cassandra23# ls -lah data/system
total 9428
...
-rw-r--r--  2 root  wheel   617B Dec 13 20:56 LocationInfo-hb-52-Data.db
-rw-r--r--  2 root  wheel    68B Dec 13 20:56 LocationInfo-hb-52-Digest.sha1
-rw-r--r--  2 root  wheel   1.9K Dec 13 20:56 LocationInfo-hb-52-Filter.db
-rw-r--r--  2 root  wheel    61B Dec 13 20:56 LocationInfo-hb-52-Index.db
-rw-r--r--  2 root  wheel   4.2K Dec 13 20:56 LocationInfo-hb-52-Statistics.db
-rw-r--r--  2 root  wheel    80B Dec 13 21:18 LocationInfo-hb-54-Data.db
-rw-r--r--  2 root  wheel    68B Dec 13 21:18 LocationInfo-hb-54-Digest.sha1
-rw-r--r--  2 root  wheel    16B Dec 13 21:18 LocationInfo-hb-54-Filter.db
-rw-r--r--  2 root  wheel    11B Dec 13 21:18 LocationInfo-hb-54-Index.db
-rw-r--r--  2 root  wheel   4.2K Dec 13 21:18 LocationInfo-hb-54-Statistics.db

Excellent, there were two. So I used sstable2json to take a look in the oldest one:

cassandra23# bin/sstable2json data/system/LocationInfo-hb-52-Data.db
{'426f6f747374726170': [['42', '01', 1323805212929000]],
 '436f6f6b696573': [['5072652d312e302068696e747320707572676564',
                     '6f68207965732c2074686579207765726520707572676564',
                     1320437246470]],
 '4c': [['436c75737465724e616d65', '737069', 1320437246450000],
        ['47656e65726174696f6e', '690e78f6', 1762556150811000],
        ['50617274696f6e6572',
         '6f72672e6170616368652e63617373616e6472612e6468742e52616e646f6d506172746974696f6e6572',
         1320437246450000],
        ['546f6b656e', '15555555555555555555555555555555', 1323805490467000]],
 '52696e67': [['00', '0a257208', 1323730075935],
              ['2aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', '0a068246', 1323730075923],
              ['40000000000000000000000000000000', '0a1d3c0e', 1323730075897],
              ['55555555555555555555555555555555', '0a25720a', 1323730075907],
              ['6aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', '0a1d3c0c', 1323730075944]]}

I’ve reformatted the output a little. The row key 4c is the “L” row from above, and the “Generation” column is the second column in the list with 47656e65726174696f6e as the column name (it’s just the hex encoded ASCII values). The column value and timestamp are the next two values.

Now the newer SSTable:

cassandra23# bin/sstable2json data/system/LocationInfo-hb-54-Data.db
{
"4c": [["47656e65726174696f6e","4ee7c086",1762556150811001,"d"]]
}

There was a single row with the 4c / “L” row key. That had a single column with “Generation” as the name, the local time of the deletion as the value, a sensible timestamp and “d” as a flag to indicate it’s a DeletedColumn.

Step 5 - Compact away the Tombstone.

It was time to purge the Tombstone by running a manual compaction on the LocationInfo CF.

cassandra23# bin/nodetool -h localhost compact system LocationInfo

Step 6 - Confirm the purge

I now expected to see a single LocationInfo SSTable:

cassandra23# ls -lah data/system/
...
-rw-r--r--  1 root  wheel    68B Dec 13 21:24 LocationInfo-hb-55-Digest.sha1
-rw-r--r--  1 root  wheel   976B Dec 13 21:24 LocationInfo-hb-55-Filter.db
-rw-r--r--  1 root  wheel    61B Dec 13 21:24 LocationInfo-hb-55-Index.db
-rw-r--r--  1 root  wheel   4.2K Dec 13 21:24 LocationInfo-hb-55-Statistics.db

In the SSTable the 4c row should be present but the 47656e65726174696f6e “Generation” column should not:

cassandra23# bin/sstable2json data/system/LocationInfo-hb-55-Data.db
{'426f6f747374726170': [['42', '01', 1323805212929000]],
 '436f6f6b696573': [['5072652d312e302068696e747320707572676564',
                     '6f68207965732c2074686579207765726520707572676564',
                     1320437246470]],
 '4c': [['436c75737465724e616d65', '737069', 1320437246450000],
        ['50617274696f6e6572',
         '6f72672e6170616368652e63617373616e6472612e6468742e52616e646f6d506172746974696f6e6572',
         1320437246450000],
        ['546f6b656e', '15555555555555555555555555555555', 1323805490467000]],
 '52696e67': [['00', '0a257208', 1323730075935],
              ['2aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', '0a068246', 1323730075923],
              ['40000000000000000000000000000000', '0a1d3c0e', 1323730075897],
              ['55555555555555555555555555555555', '0a25720a', 1323730075907],
              ['6aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', '0a1d3c0c', 1323730075944]]}

And just like that node #23 has no record of ever having a value for the “Generation” column.

Step 7 - Re-write the Generation number

The next time I restarted node #23 I needed it to get a Generation number greater than the one all the other nodes have for it. And for all restarts after that it should continue to increase the Generation.

So I set the Generation column to 1762556151 decimal or 690e78f7 hex using the cassandra-cli:

[default@system] set LocationInfo['L']['Generation'] = bytes('690e78f7');
Value inserted.
Elapsed time: 1 msec(s).

And then just to check:

[default@system] get LocationInfo['L'];                                  
=> (column=ClusterName, value=737069, timestamp=1320437246450000)
=> (column=Generation, value=690e78f7, timestamp=1323811677675000)
=> (column=Partioner, value=6f72672e6170616368652e63617373616e6472612e6468742e52616e646f6d506172746974696f6e6572, timestamp=1320437246450000)
=> (column=Token, value=15555555555555555555555555555555, timestamp=1323805490467000)
Returned 4 results.
Elapsed time: 1 msec(s).
[default@system]

Looks good, got the next Generation number in there and the timestamp is sane.

Step 8 - Confirm the Generation re-write

To get the change on disk I flushed the CF:

cassandra23# bin/nodetool -h localhost flush system LocationInfo

I now expected to see two SSTables for LocationInfo on disk:

cassandra23# ls -lah data/system/
total 9428
-rw-r--r--  1 root  wheel   588B Dec 13 21:24 LocationInfo-hb-55-Data.db
-rw-r--r--  1 root  wheel    68B Dec 13 21:24 LocationInfo-hb-55-Digest.sha1
-rw-r--r--  1 root  wheel   976B Dec 13 21:24 LocationInfo-hb-55-Filter.db
-rw-r--r--  1 root  wheel    61B Dec 13 21:24 LocationInfo-hb-55-Index.db
-rw-r--r--  1 root  wheel   4.2K Dec 13 21:24 LocationInfo-hb-55-Statistics.db
-rw-r--r--  1 root  wheel    80B Dec 13 21:32 LocationInfo-hb-57-Data.db
-rw-r--r--  1 root  wheel    68B Dec 13 21:32 LocationInfo-hb-57-Digest.sha1
-rw-r--r--  1 root  wheel    16B Dec 13 21:32 LocationInfo-hb-57-Filter.db
-rw-r--r--  1 root  wheel    11B Dec 13 21:32 LocationInfo-hb-57-Index.db
-rw-r--r--  1 root  wheel   4.2K Dec 13 21:32 LocationInfo-hb-57-Statistics.db

And if we poke into the new one:

cassandra23# bin/sstable2json data/system/LocationInfo-hb-57-Data.db
{
"4c": [["47656e65726174696f6e","690e78f7",1323811677675000]]
}

Step 9 - Restart node #23

Before the restart node #23 was using 1762556151 as the Generation. This was because the previous on disk value was 1762556150, and when it started it read this and added 1. I had now set the on disk value to be 1762556151 so after the restart I expected the Generation to be 1762556152:

cassandra23# bin/nodetool -h localhost info
Token            : 28356863910078205288614550619314017621
Gossip active    : true
Load             : 275.44 GB
Generation No    : 1762556152
Uptime (seconds) : 495
Heap Memory (MB) : 959.72 / 8032.00
Data Center      : DC1
Rack             : RAC_unknown  
Exceptions       : 0

Aces. And finally on disk the value should be 1762556152 decimal or 690E78F8 hex:

[default@system] get LocationInfo['L'];            
=> (column=ClusterName, value=737069, timestamp=1320437246450000)
=> (column=Generation, value=690e78f8, timestamp=1323812553690000)
=> (column=Partioner, value=6f72672e6170616368652e63617373616e6472612e6468742e52616e646f6d506172746974696f6e6572, timestamp=1320437246450000)
=> (column=Token, value=15555555555555555555555555555555, timestamp=1323805490467000)
Returned 4 results.
Elapsed time: 22 msec(s).

After the node came back online it started using the new Generation number and the other nodes saw it as UP.

Wrapping up

I think in theory there is a danger that if we removed node #23 from the ring and added another node with the same IP in less than 4 days it will have problems with it’s Generation not been seen as new. But thats a pretty small danger.

The important thing in all this was that the site stayed up. It’s one of the nicest things about working on Cassandra, you can fix problems while it’s working.

Oh and I fixed the other columns with wacky time stamps and am about to write a patch to log when things like this happen.

Reversed Comparators

2011-10-03T00:00:00-07:00

Cassandra 0.8.1 added support for Composite Types and Reversed Types, through a handy little type composition language. Hopefully I’ll get to show some of the things you can do with Composite Types later, for now the best resource I know is Ed Anuff’s presentation on Cassandra Indexing Techniques at Cassandra SF 2011.

The Reversed Comparator wraps another Comparator (.e.g. AsciiType) and well, there is easy way to say this, reverses the order of the comparisons so that columns are stored in descending order. If you are using something like a time stamp for a column name, the most recent value will be the first column on the row rather than the last. In a Relational DB with tree based indexes there is not much difference between getting the first or last row in an index, but Cassandra does not use tree indexes.

Recall from my post on Cassandra Query Plans that once rows get to a certain size they include an index of the columns. And that the entire index must be read whenever any part of the index needs to be used, which is the case when using a Slice Range that specifies start or reversed. So the fastest slice query to run against a row was one that retrieved the first X columns in a row by only specifying a column count.

To get an idea of how different the performance is I took the test script from Cassandra Query Plans and modified it to test two query patterns using both (regular) Ascending and (reversed) Descending comparators.

Getting the 10 most recent columns using a start column and a column count.
Getting the 10 most recent columns using only a column count.

In the second case the query would specify reversed when using (regular) Ascending ordering so Cassandra would get the last 10 columns. However when using (reversed) Descending columns it would only specify the column count.

In Motion - The Setup

The reverse_query_profile.py gist contains the Python code I used to setup the data and profile queries using the pycassa library. All tests were done locally on a 2011 Mac Book Pro with 8GB of RAM and spinning disk.

Note: you will need to edit the file and set the CASSANDRA_PATH at the top of the file.

The tests used columns with a 10 byte name and 25 bytes of data. Together with the 15 bytes of column meta data this will create columns that use 50 bytes on disk. For the standard 64KB column page this should give 1,310 columns per column page.

To test the latency of various queries I used the (recent) ‘ReadLatency: ’ metric available via nodetool cfstats. This metric wraps the local processing of a query for a single CF. The value presented by nodetool is the most recent value.

All tests were run 10 times and the min, max 80th and 95th percentiles of latency were recorded.

The Reversed Comparator is specified by using a string to specify the comparator and appending (reversed=true). I created the schema using a clean 0.8.6 install and the following bin/cassandra-cli script:

create keyspace reverse
    with strategy_options=[{replication_factor:1}]
    and placement_strategy = 'org.apache.cassandra.locator.SimpleStrategy';

use reverse;

create column family NoCache_Ascending
    with comparator = AsciiType
    and default_validation_class = AsciiType
    and key_validation_class = AsciiType
    and keys_cached = 0
    and rows_cached = 0;

create column family NoCache_Descending
    with comparator = 'AsciiType(reversed=true)'
    and default_validation_class = AsciiType
    and key_validation_class = AsciiType
    and keys_cached = 0
    and rows_cached = 0;

Note: You can use the Reversed Comparator anywhere, but it only makes sense to do so for the comparator, subcomparator and any secondary indexes.

The reverse_query_profile module inserts the following rows:

“small-row” with 100 columns, 5K of data
“no-col-index” with 1200 columns, 60K of data
“five-thousand” with 5000 columns, 244K of data
“ten-thousand” with 10000 columns, 488K of data
“hundred-thousand” with 100000 columns, 4.8M of data
“one-million” with 1000000 columns, 48M of data
“ten-million” with 10000000 columns, 480M of data

Insert the data by executing:

$ python reverse_query_profile insert_rows

To keep things simple flush and compact the database so we only have one SSTable:

$ bin/nodetool -h localhost flush
$ bin/nodetool -h localhost compact query

At the end of this my /var/lib/cassandra/data/query directory contained the following SSTables (ignoring the -Compacted SSTables):

-rw-r--r--   1 aaron  wheel   536M  3 Oct 21:49 NoCache_Ascending-g-184-Data.db
-rw-r--r--   1 aaron  wheel   2.4K  3 Oct 21:49 NoCache_Ascending-g-184-Filter.db
-rw-r--r--   1 aaron  wheel   154B  3 Oct 21:49 NoCache_Ascending-g-184-Index.db
-rw-r--r--   1 aaron  wheel   4.2K  3 Oct 21:49 NoCache_Ascending-g-184-Statistics.db
-rw-r--r--   1 aaron  wheel   536M  3 Oct 21:49 NoCache_Descending-g-149-Data.db
-rw-r--r--   1 aaron  wheel   2.4K  3 Oct 21:49 NoCache_Descending-g-149-Filter.db
-rw-r--r--   1 aaron  wheel   154B  3 Oct 21:49 NoCache_Descending-g-149-Index.db
-rw-r--r--   1 aaron  wheel   4.2K  3 Oct 21:49 NoCache_Descending-g-149-Statistics.db

reverse_query_profile contains a warm up function that will slice through all the columns in all the rows, warm up the database by executing:

$ python reverse_query_profile warm_up

In Motion - Start Column

I was not expecting much difference between the Ascending and Descending CF’s for this test as both use a start column. For the Ascending CF the script specifies the name of the tenth last column as the column name. For the Descending CF it uses the name of the last column because it will be the first column on the row and the others will follow it.

$ ./reverse_query_profile.py recent_100_start
Latency is min, 80th percentile, 95th percentile and max.
100 most recent columns, using start

Testing CF: NoCache_Ascending
Row            small-row latency in ms       0.14     0.1512     0.1542      0.157
Row         no-col-index latency in ms      0.361     0.5082     0.6311      0.751
Row        five-thousand latency in ms      0.337      0.368     0.5269      0.721
Row         ten-thousand latency in ms      0.307       0.32     0.3214      0.323
Row     hundred-thousand latency in ms      0.205     0.2362     0.2506      0.266
Row          one-million latency in ms      0.429      0.476     0.5234       0.58
Row          ten-million latency in ms      1.247      1.432       4.71       8.71
Testing CF: NoCache_Descending
Row            small-row latency in ms      0.138     0.1526     0.1697       0.19
Row         no-col-index latency in ms       0.32     0.3532     0.3572       0.36
Row        five-thousand latency in ms      0.329      0.374     0.3863        0.4
Row         ten-thousand latency in ms      0.351     0.3868     0.3875      0.388
Row     hundred-thousand latency in ms      0.357     0.4094     0.4955        0.6
Row          one-million latency in ms      0.451      0.496     0.5069      0.519
Row          ten-million latency in ms      1.312      1.401      1.415      1.424

Looks like there was a couple of outliers that mucked up the 95th percentile, but if we look at the 80th percent there is not much difference between the two CF’s.

In Motion - Count Only

This is where the differences are. When a query does not specify a start column (and does not specify reversed) the server can just start reading columns from the start without having to worry about finding the right place to start. This is exactly what we can do for the Descending CF.

For the regular Ascending CF we need to specify reversed, so the server must read the row index and work out which column is column count from the end of the row.

There is no comparison really.

$ ./reverse_query_profile.py recent_100_count
Latency is min, 80th percentile, 95th percentile and max.
100 most recent columns, using no start

Testing CF: NoCache_Ascending
Row            small-row latency in ms      0.119     0.1496     0.1577      0.167
Row         no-col-index latency in ms      0.313     0.3504     0.3879      0.433
Row        five-thousand latency in ms      0.315      0.328     0.3361      0.346
Row         ten-thousand latency in ms      0.264      0.291     0.4032      0.539
Row     hundred-thousand latency in ms      0.181     0.2038     0.2049      0.206
Row          one-million latency in ms      0.397     0.4308     0.4315      0.432
Row          ten-million latency in ms       1.29      1.421      4.538      8.341
Testing CF: NoCache_Descending
Row            small-row latency in ms      0.133     0.1508     0.1514      0.152
Row         no-col-index latency in ms      0.139      0.157     0.1633      0.171
Row        five-thousand latency in ms      0.136     0.1506     0.1532      0.156
Row         ten-thousand latency in ms      0.142     0.1558     0.1565      0.157
Row     hundred-thousand latency in ms      0.139     0.1498     0.1518      0.154
Row          one-million latency in ms      0.139     0.1592     0.1661      0.171
Row          ten-million latency in ms      0.134     0.1602     0.1678      0.175

For this type of query the Reversed Comparator provided consistent query performance no matter how many columns the row container. From now on if you are modeling time series data, such as in our old friend the Twitter clone, you should probably use the Reversed Comparator.

Token Range Ownership

2011-09-30T00:00:00-07:00

Recently, like 2 hours ago, I was planning some work to rebalance a Cassandra cluster and I wanted to see how the steps involved would effect the range ownership of the nodes. So I replicated the logic from RandomPartitioner.describeOwnership() in a handy python script.

The script is available at git hub.

In my case I had an unbalanced cluster that looked like:

Address         DC          Rack        Status State   Load            Owns    Token                                       
                                                                              154009024815050802110273337963779530663     
127.0.0.3    datacenter1 rack1       Up     Normal  430.51 GB       69.95%  	102889564695022956386161396156024583904     
127.0.0.2     datacenter1 rack1       Up     Normal  430.26 GB       22.81%   	141704132449535340642001248672108470009     
127.0.0.1    datacenter1 rack1       Up     Normal  725.61 GB       7.23%   	154009024815050802110273337963779530663

So I wanted to change the nodes to use the balanced tokens:

0
56713727820156410577229101238628035242
113427455640312821154458202477256070484

As a side affect I also want to make the token order match the node IP / host name order, which makes things a little more confusing than they need to be. So each node, together with it’s old and new token looks like:

127.0.0.1
    154009024815050802110273337963779530663
    0
127.0.0.2
    141704132449535340642001248672108470009
    56713727820156410577229101238628035242
127.0.0.3
    102889564695022956386161396156024583904
    113427455640312821154458202477256070484

To get an idea of what the cluster balance would look like as I performed the nodetool move operations I passed the initial tokens as a space separated list to the script and used the --interactive arg so I could easily enter the changes I wanted to make.

Here are the initial tokens, in the order of the host names.

./token_range.py --interactive 154009024815050802110273337963779530663 141704132449535340642001248672108470009 102889564695022956386161396156024583904
154009024815050802110273337963779530663 141704132449535340642001248672108470009 102889564695022956386161396156024583904
69.95% - 102889564695022956386161396156024583904
22.81% - 141704132449535340642001248672108470009
 7.23% - 154009024815050802110273337963779530663

First change is to wrap the token for 0.1 around to 0:

Next tokens: 0 141704132449535340642001248672108470009 102889564695022956386161396156024583904
0 141704132449535340642001248672108470009 102889564695022956386161396156024583904
16.71% - 0
60.47% - 102889564695022956386161396156024583904
22.81% - 141704132449535340642001248672108470009

Next center the middle token:

Next tokens: 0 56713727820156410577229101238628035242 102889564695022956386161396156024583904
0 56713727820156410577229101238628035242 102889564695022956386161396156024583904
39.53% - 0
33.33% - 56713727820156410577229101238628035242
27.14% - 102889564695022956386161396156024583904

Finally sort out the little piggy at the end:

Next tokens: 0 56713727820156410577229101238628035242 113427455640312821154458202477256070484
0 56713727820156410577229101238628035242 113427455640312821154458202477256070484
33.33% - 0
33.33% - 56713727820156410577229101238628035242
33.33% - 113427455640312821154458202477256070484

Note: The only testing this had was that it worked for me and matched what a live cluster was saying.

Cassandra Query Plans

2011-07-04T00:00:00-07:00

I’ve had a few conversations about query performance on wide rows recently, so it seemed about time to dig into how the different slice queries work.

Lots-o-garbage

Clearly lots of garbage in an SSTable in the form of deleted, expired, overwritten or tombstoned columns will have an impact on reads. As they increase the amount of data that will be read, examined and then discarded.

Like any database one part of query evaluation involves reading data from disk and another involves filtering the candidate data into the final result set. The filtering part of a Cassandra read query involves collating the various row fragments, reconciling multiple columns with the same name, and handling tombstones and expiring columns.

For now I’m going to focus on the reading part and I’m going to restrict it further by only considering cases where all the data is in a single SSTable and there are no garbage columns.

What’s in a row?

Aside from the columns there are a few other things stored in a row in an SSTable that are of use when performing read queries.

A row will be written to an SSTable with zero columns if it has a row level Tombstone. The Tombstone will be used when reconciling the columns for a query as it may delete columns in the row that have a lower timestamp.

Each row also contains a Bloom Filter of the column names in the row. This can be used to test if the row may contain any of the columns a query is looking for. Bloom Filters may give false positives, just like the SSTable bloom filter for row keys, so the only way to know if the column really exists is to read it.

Finally if the serialised size of the columns in the row is greater than the column_index_size_in_kb config setting (default is 64) a column index is also written. The index is a sampling of first and last column names in each column_index_size_in_kb (or more) chunk of column data, along with the offset and width of the chunk. Queries can de-serialise the index and work out how may pages of columns to skip before dropping down to the data file to scan for matching columns.

Slice By Names

A slice query that specifies a list of columns names in the SlicePredicate is turned into a SliceByNamesReadCommand internally. This is the command you will see logged by the server when DEBUG level logging is enabled. It uses the SSTableNamesIterator to read bytes off disk and create the Column objects that will be collated and reconciled to arrive at the query result.

The first thing the Iterator does is determine if the SSTable contains the row fragment we are interested in, this follows the read path described in The forces of Write and Read. When the file is opened a buffer of size column_index_size_in_kb (specified in cassandra.yaml, default is 64K) will be allocated if Standard (non memory mapped) file access is been used. If the default Memory-mapped file access is used the buffer size is ignored.

If the row exists in the SSTable the Iterator will seek to the start of the row in the Data component of the SSTable and take the following initial steps:

Read the row key.
Read the row size.
Read the row level Bloom Filter.
Read the row level column index, if present.
Read the row level Tombstone.
Filter the requested columns using the row level Bloom Filter.

At the end of these initial steps the Iterator will know if the SSTable may contain any of the columns requested in the query. Even if the SSTable table does not contain any columns of interest it may contain a row level Tombstone which will need to be reconciled with columns contained in other SSTables.

If the row may contain columns for the query the path the Iterator will take depends on presence of the column index.

If the row does not contain a column index the iterator will:

Read the number of columns in the row.
Read each column fully (including it’s value) from the SSTable.
Add the column to the result set if it’s name is in the list of columns queried for.
Stop the scan early if the number of filtered columns (from the Bloom Filter) is reached.

This is a basic scan operation over potentially all columns in the row. However the code knows it will only have to scan at most one page of columns that will contain at most column_index_size_in_kb (64K) of columns. The number of columns in the column page will vary with regard to the size of the columns, so the amount of de-serialisation and object creation will vary for each row.

When using Standard disk access with the default file buffer size is 64K the entire row may have already been read from disk. When using Memory-mapped file access the row will be backed by a MappedByteBuffer of up to 2GB. The MappedByteBuffer will use the native page size) when it asks for data, on Mac Book this is 4096 bytes (getconf PAGESIZE). The MappedByteBuffer has the advantage of not needing to copy the bytes read from disk, and it will also keep the data in memory until the OS needs to page it out.

In both cases it’s reasonable to assume there maybe no additional IO to read all the bytes for the column page.

If the row does contain a column index the Iterator will:

Use the the column index to work out the distinct ordered set of column pages that contain the columns which may exist in the row (filtered columns from the Bloom Filter).
Seek to the start position for each page of columns identified in step 1.
Scan through the entire page of columns, de-serialise each column fully (including it’s value) and add it to the result set if it’s name is in the list of columns queried for.

The search path is reduced by only scanning the pages that contain columns which passed through the bloom filter. However there is no facility to stop a particular page scan until the end of the page is reached.

For both indexed and non indexed rows all the columns which match the query are eagerly read from disk for each SSTable and held in the Iterator.

Slice From Read

A slice query that does not specify a list of columns is turned into a SliceFromReadCommand. It will contain optional start and finish column names and a column count. Again the work for reading columns from disk is done by an Iterator, this time the SSTableSliceIterator which wraps either a SimpleSliceReader or an IndexedSliceReader.

The SSTableSliceIterator is responsible for:

Ensuring the key exists in the SSTable and seeking to it’s position.
Reading the key from the row.
Reading the row size.
Deciding which Iterator to use.

The SimpleSliceReader is used when the query does not specify a start column and uses the default ascending column order (i.e. does not specify reversed). Otherwise the IndexedSliceReader is used.

When the SimpleSliceReader is used it will:

Skip and ignore the row level Bloom Filter.
Skip and ignore the column index.
Read the row level Tombstone.
Read the column count.
De-serialise each column in turn on demand as requested through the Iterator interface.

The SimpleSliceReader will only stop the scan when either the end of the row is reached, or a column with a name greater than the finish column is reached.

It’s the responsibility of the SliceQueryFilter to only request the columns it needs for the query. This involves determining which columns are live and contribute to the column count specified in the SliceRange. The query will be stopped by either:

Reading count live columns from the SSTables.
Reading past the finish column if specified.
Exhausting all SSTables involved in the read.

The IndexedSliceReader has a much tougher life. It is used when there is a start column or the columns must be returned in reverse order. It will:

Skip and ignore the row level Bloom Filter.
Read the column index.
De-serialise the row and read the row level tombstone if present.

Depending on the presence of the column index one of two algorithms will be used to read the columns from disk. The SimpleBlockFetcher is used when the row does not contain a column index. It de-serialises the columns in the row, and builds a list that is either disk ordered or reverse disk ordered depending on the query. The scan is stopped by either:

Reading all columns in the row.
Reading a column beyond the finish column, if a finish column was specified and reverse is false.
Reading a column beyond the start column, if a start column was specified and reverse is true.

Unlike the SimpleSliceReader when the IndexedSliceReader is used on a small row it will eagerly read all columns from the row which match the query. This only happens in the context of a row which has less then column_index_size_in_kb of column data .

The IndexedBlockFetcher is used for all other cases, it will:

Use the column index to find the first column page to read from.
Step forwards or backwards (for reversed) through the column pages for the row on demand as each page of columns is exhausted.
Stop the iteration if the IndexInfo for the next column page does not contain columns inside the start or finish column range, if a start or finish column are specified.
De-serialise all columns in each page, building a list of columns in either disk order or reverse disk order.
Stop a page scan using the same criteria as the SimpleBlockFetcher above.

The IndexedBlockFetcher greedily reads the columns in each page just as the SimpleBlockFetcher does. But will only advance to the next page when the query requests more columns and the previous block of columns has been exhausted.

So What ?

Not much.

Cassandra works like other databases by using a seek and partial scan approach to finding data. Read requests that require fewer seek+scan operations will be faster than those the require more.

There are a couple of things we can say about what will make some queries go faster than others though.

Name Locality

Queries by column name that select columns on fewer column pages should be faster that those which are spread out over more column pages. Querying more column pages means reading more data and creating more Column objects. The pathological case is selecting n columns from n column pages.

Note that queries by column name must also de-serialise the column index, and will pay a constant cost for every query regardless of the number of columns requested or their distribution in the row.

Start Position

The entire column index must be de-serialised whenever an offset into the row needs to be found. Queries by that slice columns without specifying a start column and use (default) ascending order will perform better than those that either use a start column or reverse order. However the cost of de-serialising the column index is constant and the position of the start column should have little affect on query performance.

The overhead should naturally increase with the width of the rows.

In Motion - The Setup

The query_profile gist contains the Python code I used to setup the data and profile queries using the pycassa library. All tests were done locally on a 2011 Mac Book Pro with 8GB of RAM and spinning disk.

Note: you will need to edit the file and set the CASSANDRA_PATH at the top of the file.

To verify the column paging I recompiled the 0.8.1 source to include a change to the SSTableNamesIterator that logged the number of index entries each row had and how many column pages were used during a read.

The output made the paging look “good enough”, for example:

#selecting from 1 page
INFO 13:21:52,662 Column Page Count 4/1
INFO 13:22:03,309 Column Page Count 8/1
INFO 13:22:14,070 Column Page Count 77/1
INFO 13:22:26,140 Column Page Count 763/1
INFO 13:22:49,362 Column Page Count 7628/1

#selecting from 50 pages 
INFO 13:32:15,501 Column Page Count 4/4
INFO 13:32:25,696 Column Page Count 8/8
INFO 13:32:36,790 Column Page Count 77/50
INFO 13:32:54,012 Column Page Count 763/50
INFO 13:33:37,511 Column Page Count 7628/50

#sometimes it was off
INFO 13:35:02,987 Column Page Count 77/53
INFO 13:35:04,153 Column Page Count 77/50
INFO 13:35:05,325 Column Page Count 77/51
INFO 13:35:06,489 Column Page Count 77/52
INFO 13:35:15,812 Column Page Count 763/71
INFO 13:35:18,182 Column Page Count 763/63
INFO 13:35:20,536 Column Page Count 763/69
INFO 13:35:22,898 Column Page Count 763/70
INFO 13:35:51,997 Column Page Count 7628/67
INFO 13:36:07,069 Column Page Count 7628/65
INFO 13:36:22,012 Column Page Count 7628/66

All tests were run 10 times and the min, max 80th and 95th percentiles of latency were recorded.

Using a clean 0.8.1 install create the following keyspace using bin/cassandra-cli:

create keyspace query
    with strategy_options=[{replication_factor:1}]
    and placement_strategy = 'org.apache.cassandra.locator.SimpleStrategy';

use query;

create column family NoCache
    with comparator = AsciiType
    and default_validation_class = AsciiType
    and key_validation_class = AsciiType
    and keys_cached = 0
    and rows_cached = 0;

The query_profile module will insert the following rows:

“small-row” with 100 columns, 5K of data
“no-col-index” with 1200 columns, 60K of data
“five-thousand” with 5000 columns, 244K of data
“ten-thousand” with 10000 columns, 488K of data
“hundred-thousand” with 100000 columns, 4.8M of data
“one-million” with 1000000 columns, 48M of data
“ten-million” with 10000000 columns, 480M of data

Insert the data by executing:

$ python query_profile insert_rows

To keep things simple flush and compact the database so we only have one SSTable:

$ bin/nodetool -h localhost flush
$ bin/nodetool -h localhost compact query

At the end of this my /var/lib/cassandra/data/query directory contained the following SSTables (ignoring the -Compacted SSTables):

-rw-r--r--  1 aaron  wheel   536M  6 Jul 13:03 NoCache-g-80-Data.db
-rw-r--r--  1 aaron  wheel   1.4K  6 Jul 13:03 NoCache-g-80-Filter.db
-rw-r--r--  1 aaron  wheel   154B  6 Jul 13:03 NoCache-g-80-Index.db
-rw-r--r--  1 aaron  wheel   4.2K  6 Jul 13:03 NoCache-g-80-Statistics.db

query_profile contains a warm up function that will slice through all the columns in all the rows, warm up the database by executing:

$ python query_profile warm_up

In Motion - Name Locality

We want to test that selecting columns by name when they are tightly grouped has better performance than selecting widely distributed columns.

The name_locality test runs through several tests that select up to 100 columns by name from each of the rows. The numbers show a reasonably stable latency for queries that touch the same number of pages, and an increase when the number of pages increases. Which matches the theory.

Note: the code outputs a WARN message when the test selects less than 100 columns from a row. For various tests this happened on the “small-row”, “no-col-index”, “five-thousand” and “ten-thousand” rows. Mostly when selecting a set number of columns from each column page.

$ python query_profile.py name_locality
Latency is min, 80th percentile, 95th percentile and max.
Test name locality...

100 columns by name, start of the row.
Row            small-row had latency in ms      0.285      0.362     0.3828      0.396
Row         no-col-index had latency in ms      0.848     0.9146     0.9159      0.917
Row        five-thousand had latency in ms      1.257      1.321      1.324      1.324
Row         ten-thousand had latency in ms      1.226      1.311      1.333      1.358
Row     hundred-thousand had latency in ms      1.406      1.483      1.489      1.492
Row          one-million had latency in ms      2.848      2.957      3.019      3.089
Row          ten-million had latency in ms      17.75      18.72      18.76      18.78

100 columns by name, end of the row.
Row            small-row had latency in ms      0.273     0.2846     0.2891      0.294
Row         no-col-index had latency in ms      0.674     0.7078     0.7107      0.714
Row        five-thousand had latency in ms      0.928      0.962     0.9638      0.966
Row         ten-thousand had latency in ms      0.882      0.901     0.9172      0.937
Row     hundred-thousand had latency in ms      0.868     0.9138     0.9154      0.917
Row          one-million had latency in ms      2.553       2.72       2.73      2.739
Row          ten-million had latency in ms      16.95      17.67      17.82      17.97

100 columns by name, middle of row.
Row            small-row had latency in ms      0.273     0.3134     0.3422      0.368
Row         no-col-index had latency in ms      0.835     0.9052     0.9092      0.913
Row        five-thousand had latency in ms      1.719      1.772      1.839      1.915
Row         ten-thousand had latency in ms      1.726      1.802      1.818      1.837
Row     hundred-thousand had latency in ms      1.853      1.978      6.987       13.1
Row          one-million had latency in ms      2.682      2.832      2.927      3.041
Row          ten-million had latency in ms      17.27      18.59      18.72      18.75

100 columns by name, first 2 cols from 50 random pages
Row            small-row had latency in ms      0.115      0.119     0.1195       0.12
Row         no-col-index had latency in ms       0.37     0.4008     0.4019      0.403
Row        five-thousand had latency in ms      1.504      1.596      1.614      1.637
Row         ten-thousand had latency in ms       3.14      3.445      3.493      3.547
Row     hundred-thousand had latency in ms      25.25      27.63      36.62       47.5
Row          one-million had latency in ms      26.68      28.47      28.92      29.47
Row          ten-million had latency in ms      40.95      43.55      43.96      44.13

100 columns by name, last 2 cols from 50 random pages
Row            small-row had latency in ms      0.109      0.115     0.1218       0.13
Row         no-col-index had latency in ms      0.349     0.3536     0.3553      0.357
Row        five-thousand had latency in ms      1.778      1.854      1.877      1.897
Row         ten-thousand had latency in ms       3.39      3.664      3.715      3.765
Row     hundred-thousand had latency in ms      25.17      26.77       28.0      29.35
Row          one-million had latency in ms      26.44      28.04      28.19      28.35
Row          ten-million had latency in ms      40.57      44.26      44.69      44.74

100 columns by name, random 2 cols from 50 random pages
Row            small-row had latency in ms      0.107      0.117     0.1174      0.118
Row         no-col-index had latency in ms       0.36     0.4356     0.4445      0.445
Row        five-thousand had latency in ms       1.72      1.808      2.795      3.998
Row         ten-thousand had latency in ms      3.367      3.682      3.705      3.715
Row     hundred-thousand had latency in ms      25.45      27.23      28.93      30.95
Row          one-million had latency in ms       32.6      38.43      38.91      39.45
Row          ten-million had latency in ms      48.76      53.88      54.09      54.19

100 columns by name, first col from 100 random pages
Row            small-row had latency in ms      0.106      0.113     0.1306      0.152
Row         no-col-index had latency in ms      0.326     0.3508      0.351      0.351
Row        five-thousand had latency in ms      1.361      1.451      1.483      1.518
Row         ten-thousand had latency in ms      2.853      3.071      3.102      3.136
Row     hundred-thousand had latency in ms      37.65      40.08      40.92      41.84
Row          one-million had latency in ms      51.23      54.41      55.93       57.6
Row          ten-million had latency in ms      64.67      68.56      69.24       69.6

100 columns by name, last col from 100 random pages
Row            small-row had latency in ms      0.105      0.113     0.1153      0.118
Row         no-col-index had latency in ms      0.325      0.349     0.3553      0.363
Row        five-thousand had latency in ms      1.601      1.695      1.698        1.7
Row         ten-thousand had latency in ms      3.058      3.404      4.317      5.429
Row     hundred-thousand had latency in ms      37.34      39.52      40.54      41.72
Row          one-million had latency in ms      50.76      53.92      54.58      55.14
Row          ten-million had latency in ms       64.2      68.46      69.28      70.07

100 columns by name, random col from 100 random pages
Row            small-row had latency in ms       0.11     0.1156     0.1169      0.118
Row         no-col-index had latency in ms      0.327     0.3628     0.3721      0.382
Row        five-thousand had latency in ms      1.541      1.656      1.664      1.674
Row         ten-thousand had latency in ms      2.719      3.272      3.279      3.282
Row     hundred-thousand had latency in ms      35.51      37.52       38.8      40.35
Row          one-million had latency in ms      50.11      52.91       53.8       54.7
Row          ten-million had latency in ms      65.06      68.67       69.3      69.88

In Motion - Start Position

We want to show that any start position in a (forward) slice query has worse performance than one that does not. And that the position of the start column has little impact on the query performance.

The start_position test runs through several tests that select a slice of up to 100 columns with different start positions.

The test for “100 columns from the start of the row with a start col” takes significantly longer than “100 columns from with no start column” for all rows other than “small-row”. For all rows other than “no-col-index” I attribute this to the column index. For “no-col-index” row I attribute the increased latency to eagerly reading the entire page of columns. Performance was then reasonably similar for the other tests which used a start column.

I’m not sure why the test for “100 columns from the start of the second page” was an outlier. Hopefully I’ll get to take a longer look, but it seemed to consistency perform worse than other tests.

$ python query_profile.py start_position
Test start position...

Latency is min, 80th percentile, 95th percentile and max.
100 columns from with no start column
Row            small-row latency in ms      0.181     0.1934     0.2134      0.237
Row         no-col-index latency in ms       0.18     0.1918     0.1956        0.2
Row        five-thousand latency in ms      0.169      0.185     0.1864      0.188
Row         ten-thousand latency in ms      0.187     0.1936     0.1972      0.201
Row     hundred-thousand latency in ms      0.183     0.1936     0.1963      0.199
Row          one-million latency in ms      0.181     0.1948     0.1959      0.197
Row          ten-million latency in ms      0.178     0.1952     0.1965      0.197

Latency is min, 80th percentile, 95th percentile and max.
100 columns from the start of the row with a start col
Row            small-row latency in ms      0.135      0.142     0.1424      0.143
Row         no-col-index latency in ms       0.34     0.3498       0.35       0.35
Row        five-thousand latency in ms      0.358     0.3676     0.3684      0.369
Row         ten-thousand latency in ms       0.36      0.376      0.511      0.676
Row     hundred-thousand latency in ms      0.358        0.4     0.5005      0.622
Row          one-million latency in ms      0.456      0.495     0.5006      0.505
Row          ten-million latency in ms      1.345      1.418       1.44      1.467

Latency is min, 80th percentile, 95th percentile and max.
100 columns from the start of the second page
Row            small-row latency in ms      0.128     0.1412     0.1424      0.143
Row         no-col-index latency in ms      0.327      0.338     0.3385      0.339
Row        five-thousand latency in ms      0.637     0.6488     0.6634      0.681
Row         ten-thousand latency in ms      0.615      0.651     0.6519      0.653
Row     hundred-thousand latency in ms      0.628     0.6726     0.6762       0.68
Row          one-million latency in ms      0.706     0.7636     0.7671      0.771
Row          ten-million latency in ms      1.569      1.706      3.148       4.91

Latency is min, 80th percentile, 95th percentile and max.
100 columns starting half way through the row
Row            small-row latency in ms      0.099     0.1028     0.1125      0.124
Row         no-col-index latency in ms      0.354     0.3626      0.363      0.363
Row        five-thousand latency in ms      0.397     0.4068     0.4318      0.462
Row         ten-thousand latency in ms      0.389     0.4138     0.4167       0.42
Row     hundred-thousand latency in ms      0.358     0.3828      0.383      0.383
Row          one-million latency in ms      0.456     0.5048     0.5135      0.524
Row          ten-million latency in ms      1.347       1.47      1.489      1.504

Latency is min, 80th percentile, 95th percentile and max.
100 columns starting from the last page 
Row            small-row latency in ms      0.126      0.143     0.1434      0.144
Row         no-col-index latency in ms      0.325     0.3406     0.3425      0.343
Row        five-thousand latency in ms      0.595      0.634     0.6646      0.696
Row         ten-thousand latency in ms      0.556     0.5694     0.5795      0.591
Row     hundred-thousand latency in ms      0.441     0.4942     0.5027      0.512
Row          one-million latency in ms      0.405     0.4468     0.4507      0.454
Row          ten-million latency in ms      1.272      1.372      1.393      1.417

Down For Me?

2011-06-13T00:00:00-07:00

For a read or write request to start in Cassandra at least as many nodes must be seen as UP by the coordinator node as the request has specified via the ConsistencyLevel. Otherwise the client will get an UnavailableException and the cluster will appear down for that request. That may not necessarily mean it is down for all keys or all requests.

The Replication Factor, number of nodes, the Consistency Level and luck all play a part of determining how many nodes can be lost in a Cassandra cluster before it is unavailable for 100% of the keys. Before it reaches that point though the cluster may go through a period of partial failure where some keys will not be available at some CL levels.

The partial failure support baked into the system is nice thing to have. But most people will be interested in keeping 100% of the keys available at the required Consistency Level. So most of discussion below talks about keeping the cluster up for 100% of the keys at the QUORUM CL.

Which nodes ?

When it comes to counting the UP nodes for a request we only consider the Natural Endpoints for a key. These are the nodes identified by the placement_strategy (set when the Keyspace was created) as the replicas for a key, and they never change. All read and write operations for a key, using the same partitioner, will select those same endpoints. Otherwise write operations could plonk down data that reads could never find.

The row key is first Decorated by the partitioner (specified in conf/cassandra.yaml) to create the token used to locate the row in the cluster. For example the RandomPartitioner uses an MD5 transform to turn the key into a 128bit token.

Is using the SimpleStrategy it will:

Order the nodes in the ring by their initial_token.
Select the node whose token range includes the token as the first replica.
Select the next RF-1 nodes in the ordered ring as the remaining replicas.

The nodes loop around, so a row may be replicated on the last 2 nodes and the first 1.

The NetworkTopologyStrategy use a more involved algorithm that considers the Data Centre and Rack the node is assigned to by the Snitch. There is a good discussion of the process from Peter Schuller.

The weird Consistency Level

There is one Consistency Level that does not behave like the others, so lets just get it out of the way first. CL ANY for write requests allows the coordinator node to store the mutation in the form of a Hinted Handoff on any node in the cluster, which in practice means on the coordinator itself.

This is useful in cases where extreme write uptime is needed. The sort of extreme where the write cannot be reliably read until a nodetool repair operation has been completed. Hinted Handoffs must be delivered to Natural Endpoints before they can be included in a read operation.

If you write at CL ANY and some of the Natural Endpoints are up, the write and the Hints will be sent to them. The coordinator will only be used to store the Hints in cases where all the Natural Endpoints are down.

For more information on Hinted Handoff see Jonathan’s recent post.

Consistency Levels

For all the other Consistency Levels the read or write request is directed to one of the Natural Endpoints. Hinted Handoffs may be used as part of the request but are do not considered when determining if the cluster is available for the request.

The named Consistency Levels ONE, TWO and THREE are pretty easy to understand. Once, two or three replicas for the key must be seen as UP before the operation will start. CL ONE is the most often used of these.

QUORUM is calculated as floor(RF \ 2) + 1. This is most used CL level and in my opinion should be the starting CL for all applications until a reason is found to change (performance is not a reason).

For RF levels below 3 the QUORUM is the same as the RF level so:

RF 1 - QUORUM = 1
RF 2 - QUORUM = 2
RF 3 - QUORUM = 2
RF 4 - QUORUM = 3
RF 5 - QUORUM = 3
RF 6 - QUORUM = 4
RF 7 - QUORUM = 4

When the NetworkTopologyStrategy is used each data centre has it’s own RF, and the standard QUORUM is calculated using the total RF for the cluster.

LOCAL_QUORUM and EACH_QUORUM can be used with the NetworkTopologyStrategy and they instruct the coordinator to also consider the Data Centre the nodes are located in. A write is always sent to all UP replicas, this

For LOCAL_QUORUM only the RF of the local data centre is considered when:

Calculating how many nodes to block for.
Checking if enough nodes are UP for the request.
Counting if CL nodes have responded to the request.

EACH_QUORUM works in a similar way but the tests apply to every DC in the cluster.

ALL requires that all replicas for the row be UP before the request will start.

A failing Range

A simple, but incomplete, way to think about the cluster been available is to focus a one key range and it’s replicas.

Consider RF 3, at QUORUM if one node is lost the range will still be available. If two nodes are lost the range will not be available for QUORUM operations, but will still be available for ONE and ANY requests.

For any number of nodes in the cluster, a range will become unavailable it more than (RF - CL) nodes are DOWN.

Now consider a cluster with 50 nodes, RF 5, QUORUM operations and the SimpleStrategy. If 3 adjacent nodes nodes go down the range assigned to the first one will no longer be available, as their will only be 2 replicas UP. The cluster will be down for 2% of the possible keys, not terrible but it’s no longer up for 100%. The range assigned to the second down node will have 3 UP replicas and the range assigned to the third will have 4 UP replicas. The nodes do not have to be adjacent in the ring for this occur, it could be any nodes in the replica set for the range. It’s just easier to think about when they are adjacent.

Spreading replicas for a key across nodes with different physical infrastructure is a good way to mitigate this risk. The NetworkTopologyStrategy distributes the replicas for a DC across the available Racks. As defined by either the RackInferringSnitch, PropertyFileSnitch or the EC2Snitch which uses AWS Availability Zones as racks. The SimpleSnitch puts all nodes into rack1 in datacenter1.

In the worst case failure scenario the cluster can sustain up to (RF - CL) failures and still remain available for 100% of the keys.

A failing Cluster

The best case scenario for failure is when the node failures are evenly divided amongst the replicas for a range. So that every RF number of failures only removes one node from the available replica set for each range. To know how many nodes we can lose for a Consistency Level to still be available for 100% of the keys multiply by RF-CL.

For a 5 node cluster with RF 3 at QUORUM this is (5 / 3) * (3 - 2) or 1. For other cluster sizes the number is:

def max_failure(num_nodes, rf, block_for):
    print "For up to %s nodes with RF %s and blocking for %s nodes..." % (
        num_nodes, rf, block_for)
    print "Number Nodes / max_failure";
    for n in range(1, num_nodes + 1):
         print "%s / %s" % (n, ( int(n/rf) * (rf - block_for)))

max_failure(10, 3, 2)

For up to 10 nodes with RF 3 and blocking for 2 nodes...
Number Nodes / max_failure
1 / 0
2 / 0
3 / 1
4 / 1
5 / 1
6 / 2
7 / 2
8 / 2
9 / 3
10 / 3

In the best case failure scenario the cluster can sustain up to floor(number of nodes / RF) * (RF - CL) failed nodes and still remain up for 100% of the keys.

The view from one node

Considering which nodes are UP or DOWN is always from the perspective of the coordinator node. Network Partitions also play a part in the deciding if the cluster is available, as the nodes must be both running and contactable by the coordinator.

At the small scale if a client connects to a node in the cluster that has lost connectivity to other nodes in the cluster it will consider them all DOWN and be unavailable for all QUORUM requests. The client will receive a UnavailableException and should connect to another node and try request. Other nodes in the cluster may be in a bigger partition that contains enough UP replicas for the request to complete.

At a bigger scale when using Amazon AWS it’s more likely that nodes an Availability Zone will lose connectivity with nodes from a different AZ then from nodes in the same AZ.

With two AZ’s operations at QUORUM will require nodes in both AZ’s as neither will hold QUORUM replicas. So a network partition between the two would result in 100% of the keys been down in both AZ’s.

With three AZ’s each AZ will hold one third of the replicas, and any two together may provide enough UP replicas to support QUORUM operations. The cluster could sustain a network partition so long as each AZ can talk to at least one other AZ.

Recent Cassandra AWS Reading

2011-05-21T00:00:00-07:00

There’s been a few good AWS discussions recently on the Cassandra User List, and some interesting blog posts.

User List discussions

best way to backup on EC2 and the SimpleGeo.com table snap
EC2 Stress Tests with lots of hard numbers
Multi DC Deployment discussion
Work on a new EC2 Snitch to make multi region deployments easier using CASSANDRA-2452
Using Vyatta and IP address resolution in MultiDC setup (EC2)/VIP
Token selection for a balanced Replica data distributing between racks

Blogs and such

Setting up a Cassandra cluster with the DataStax AMI
Installing the Brisk AMI on Amazon EC2
Amazon documentation on EC2 Instance Storage
Persistence Strategies for Amazon EC2
Summary of the Amazon EC2 and Amazon RDS Service Disruption in the US East Region
RAID0 ephemeral storage on AWS EC2
Three Latency Anomalies a look at EC2 networking

Deletes and Tombstones

2011-05-15T00:00:00-07:00

Deletes in Cassandra rely on Tombstones to support the Eventual Consistency model. Tombstones are markers that can exist at different levels of the data model and let the cluster know that a delete was recored on a replica, and when it happened. Tombstones then play a role in keeping deleted data hidden and help with freeing space used by deleted columns on disk.

It’s possible to delete columns, super columns and entire rows in Cassandra. However in this post I’m going to look at the simple case of deleting a single column in a Standard Column Family.

Remembering things forgotten

In a simple case of a single node traditional RDBMS deleting data is relatively straight forward process. Rows in Index and Data pages are marked as unused and other than recording the transaction in the commit log, the fact that the delete happened is forgotten.

As the Distributed Deletes wiki page points out, things are a bit more complicated in Cassandra. From the perspective of the coordinator node (the one the client is connected to) one or more nodes may be down when the delete is executed. So long as the requested Consistency Level is achieved the delete can still proceed, but if we forget that the delete happened it will not be possible to reach the correct consistent view of the data later. The nodes that were offline will say “foo == bar” and the nodes that did the delete will have nothing to say.

When a column is deleted a DeletedColumn aka Tombstone is created in Cassandra. The DeletedColumn will have:

name: Name of the column deleted
value: Current server time as seconds since the unix epoch (integer). This is known as the localDeleteTime and is used during the (cassandra) GC process.
timestamp: As provided by the client

The mutation is then applied to the memtable in one of two ways. If the memtable does not contain the named column for the row it is simply added to the memtable. If there is an existing column it is reconcile() ‘d with the Deleted Column. The new DeletedColumn will replace the existing column if it has a higher (client provided) timestamp. The localDeleteTime is not used for reconciliation. At this point any previous column value in the memtable is lost and will not be persisted to disk.

We now have a tombstone. If there are no other mutations the DeletedColumn will later be persisted to the SSTable just like any other column.

Local Reads for Local Queries

During a local read for a row value the same reconciliation process that was used during the delete request runs. Multiple row fragments are retrieved from the current memtable, memtables pending flush and SSTables on disk. The fragments are reduced and the columns with the same name reconciled to arrive at the current value.

For example if there is a row fragment in an SSTable for key “foo” that says columns “bar” is “baz”, and a DeletedColumn in another SSTable with a higher time stamp when they are reconciled the DeletedColumn will “win”. The current view of the row will be that the “bar” column is deleted.

When a read query is filtering the reconciled candidate columns it will include the DeletedColumn in the result set if the localDeletionTime recorded for it is beyond the current gcBefore time. gcBefore is determined when the query starts as the current time (to 1 second resolution) less the GCGraceSeconds value specified for the Column Family. If the deletion is before gcBefore it is totally ignored, more on the GC process below. When a slice based query (e.g. get the first 10 columns) executes it adds the DeletedColumn to the result set, but does not count it towards the limit of columns the query has asked for.

The deleted columns are filtered out of the result set as the last step before returning it to the client. In the simple case of a local read at CL ONE the result of the local query is filtered and returned. For cluster reads the result of the query must be reconciled with CL replicas before the coordinator can say it has a consistent result. And this is why the DeletedColumn’s are still in the result set.

Readers Digest

If a read request involves more than one replica only the “closest” (as determined by the snitch) is asked to return the full result set. The result set it returns is will include any DeletedColumn’s read during the local read.

The other nodes in the cluster are asked to return digest of their local read. The digest is a MD5 hash of the columns, their values, timestamps and other meta data. Once the coordinator node has received CL read responses, including the data response, it compares the digests with each other and the digest of the data response.

If they all match then the read is consistent at the CL requested. Otherwise there is an inconsistency that needs to be repaired. The inconsistency could come from one replica including a DeletedColumn in it’s digest while another includes the previously deleted value.

Read Repair (sort of)

Once a DigestMismatch is detected the differences have to be reconciled before the read response is returned to the client. The process that does this is part of the ReadRepair feature, but depending on the circumstances it may not be considered a full Read Repair.

Read Repair is considered to be happening when the coordinator requests a response from all replicas for a row, no just those that are needed to meet the Consistency Level for the request. This read repair will run for normal get / multi_get / indexed get operations but not range scans, and can be controlled with the read_repair_chance config setting. Even though all replicas receive the request, the request will only block of Consistency Level replicas. For now I’m going to ignore this process and just consider what happens if replicas required for the Consistency Level do not agree.

Once the DigestMismatch is detected all the replicas that were involved in the read are asked to do the read again and return a full data response to the coordinator. Their responses are then reconciled using the same process as a normal mutation request to get a consistent result for the query.

This is where the tombstone do their work of remembering that the delete happened. For example if replica 1 says ‘foo’ is a normal column, and replica 2 says ‘foo’ is a DeletedColumn with a higher time stamp the value from replica 2 will be used.

Once a reconciled view of the row has been created each replica is asynchronously sent a mutation with the difference between it’s data and the reconciled data. The read can then return the reconciled view of the data to the client while the repair of the replicas that participated in the request is going on in the background.

Free

Existing data on disk for a column is not deleted when the delete mutation is processed. Cassandra never mutates on disk data.

Instead the compaction process reconciles the data in multiple SSTables on disk. The row fragments from each SSTable are collated and columns with the same name reconciled using the process we’ve already seen. The result of the compaction is a single SSTable that contains the same “truth” as the input files, but may be considerably smaller due to reconciling overwrites and deletions.

For example, there could be three SSTables that contain a value for the “foo” column. In the first the value is “bar”, in the second the value is a 16KB string, and in the third it’s a DeletedColumn. Before compaction runs the value of the column is nothing, however on disk it uses at least 16KB. After compaction the value will still be nothing, but it will be stored in a single SSTable and use only a few bytes.

Minor compaction typically runs frequently, so data that is created and deleted reasonably quickly will be deleted from disk quickly. Data that has been through several generations of compaction before it is deleted will not be deleted from disk as quickly. The DeletedColumn will continue to be written into the new, compacted, SStables until it’s localDeletionTime occurs before the current (server) time less the GCGraceSeconds.

The reincarnation of Paul Revere’s horse

Compactions run locally and by default automatically based on load. Once GCGraceSeconds has elapsed since the delete a new compaction on the SSTable will purge the tombstone from disk and the delete will be forgotten. But how do we guarantee that other nodes in the cluster have seen the delete before it’s deleted?

If a node goes down for longer than max_hint_window_in_ms it will no longer have hints recorded for it. If the column was never read a Read Repair could not have run. If the column was read but Read Repair was not active and the node was not included in CL nodes it would not have received a repair.

Deletes operate under Eventual Consistency just like writing a value, with the added complication that they have an built in expiry time (GCGraceSeconds). If the replicas for a value have not seen the delete before that time there is a risk of deleted data reappearing. The stop that happening nodetool reapir needs to be run at least every GCGraceSeconds.

In Motion on a single node

There is not a lot to look at on a single node, but it’s pretty easy to see the tombstones and the column value persisted into an SSTable.

One a fresh 0.7 install create the sample schema:

$ bin/cassandra-cli -h localhost -f conf/schema-sample.txt

The jump into the cassandra-cli and insert one column:

$ bin/cassandra-cli -h localhost
Connected to: "Test Cluster" on localhost/9160
Welcome to cassandra CLI.

Type 'help;' or '?' for help. Type 'quit;' or 'exit;' to quit.
[default@unknown] use Keyspace1;                                    
Authenticated to keyspace: Keyspace1
[default@Keyspace1] set Standard1['foo']['bar'] = 'baz';

Flush the data from the memtable so our delete cannot be applied in memory:

$ bin/nodetool -h localhost flush Keyspace1

Now delete the column using the cli:

[default@Keyspace1] del Standard1['foo']['bar'];        
column removed.

Flush the delete to disk:

$ bin/nodetool -h localhost flush Keyspace1

Finally use the sstable2json dump the data from the SSTables:

$ bin/sstable2json /var/lib/cassandra/data/Keyspace1/Standard1-f-1-Data.db 
{
"666f6f": [["626172", "62617a", 1305412876934000, false]]
}
$ bin/sstable2json /var/lib/cassandra/data/Keyspace1/Standard1-f-2-Data.db 
{
"666f6f": [["626172", "4dcf05ab", 1305413035092000, true]]
}

The top level key in the output is the row key and each column is formatted as: name, value, timestamp, delete flag. The first SSTable still contains the “bar” column and the “baz” data written in the first set operation. The second SSTable also has the “bar” column, however the deleted flag is true and the value is now the (server) time stamp of deletion. This is the timestamp used with GCGraceSeconds.

Read the data back:

[default@Keyspace1] get Standard1['foo']['bar'];
Value was not found

In Motion on a cluster

It’s a bit more involved but it’s also possible to see a Read Repair happening on a cluster.

I normally run a 2 node cluster on my mac book using the direction from Gary Dusbabek. There is also this really handy tool from Sylvain that I’ve been meaning to try, or you can just run a normal two node cluster.

Edit conf/cassandra.yaml for both nodes to disable Hinted Handoff:

# See http://wiki.apache.org/cassandra/HintedHandoff
hinted_handoff_enabled: true

This will prevent the nodes telling each other about mutations that happen while the other is down.

I’ve also edited conf/log4j-server.properties to set logging at DEBUG.

Start the nodes and create and populate schema below using bin/cassandra-cli (schema definition is for a 0.7.5 cluster):

create keyspace dev 
  with placement_strategy = 'org.apache.cassandra.locator.SimpleStrategy'
  and replication_factor = 2;

use dev;

create column family data 
  with comparator = AsciiType;

set data[ascii('foo')]['bar'] = 'baz';

Flush the data on both nodes:

$ bin/nodetool -h your-node1 flush dev
$ bin/nodetool -h your-node2 flush dev

Shutdown node 1, connect to node 2 and delete the column using the cli:

$ bin/cassandra-cli -h your-node2
Connected to: "Test Cluster" on 127.0.0.2/9160
Welcome to cassandra CLI.

Type 'help;' or '?' for help. Type 'quit;' or 'exit;' to quit.
[default@unknown] use dev;
Authenticated to keyspace: dev
[default@dev] del data['foo']['bar'];
column removed.

Restart node 1, connect via the cli, set the consistency level to ALL to ensure Read Repair runs and read the deleted column:

$ bin/cassandra-cli -h your-node1
Connected to: "Test Cluster" on 127.0.0.1/9160
Welcome to cassandra CLI.

Type 'help;' or '?' for help. Type 'quit;' or 'exit;' to quit.
[default@unknown] use dev;
Authenticated to keyspace: dev
[default@dev] consistencylevel as ALL;
Consistency level is set to 'ALL'.
[default@dev] get data['foo']['bar']; 
Value was not found

This would have also worked at QUORUM (which is 2 for RF 2).

Digging through the logs on node1 you should see that the read will block for 2 nodes and Read Repair is enabled:

ReadCallback.java (line 84) Blockfor/repair is 2/true; setting up requests
    to localhost/127.0.0.1,/127.0.0.2

Next should be some messages about reading the data locally and handling the response from your-node2. Then the RowDigestResolver will say it’s resolving 2 responses before detecting the mismatch and raising an error:

RowDigestResolver.java (line 62) resolving 2 responses
StorageProxy.java (line 398) Digest mismatch:
    org.apache.cassandra.service.DigestMismatchException: Mismatch for key
        DecoratedKey(110673303387115207421586718101067225896, 666f6f)
        (34ec2eb2ec21eb3d05fb6f97cbf84c51 vs ba720207d87132da833ae2579487b172)
    at org.apache.cassandra.service.RowDigestResolver.resolve(
        RowDigestResolver.java:106)
    at org.apache.cassandra.service.RowDigestResolver.resolve(
        RowDigestResolver.java:30)
...

There should now be a few messages about doing the read again, and receiving the response from your-node2 again. Once the data responses have been received by node1 it will resolve the differences and send out mutations to the nodes than need them. In this case it’s node1 and you can see the mutation logged from Table (aka Keyspace):

RowRepairResolver.java (line 50) resolving 2 responses
RowRepairResolver.java (line 76) versions merged
RowRepairResolver.java (line 85) resolve: 1 ms.
Table.java (line 337) applying mutation of row 666f6f

If you run the read again you should see the RowRepairResolver log ‘digests verified’ to say the data matched and there are no repairs to run.

How are Memtables measured?

2011-05-04T00:00:00-07:00

Updated: I’ve added information on the new memtable_total_space_in_mb setting in version 0.8 and improved the information about memtable_throughput. Thanks for the feedback.

In version 0.7 of Cassandra there are 3 configuration settings that can trigger flushing a memtable to disk. Version 0.8 adds support for a global memtable_total_space_in_mb which may replace the previous 3 settings.

First the 0.7 settings.

memtable_flush_after (minutes)

This is the maximum number of minutes a memtable should stay in memory for if it has received writes. When the memtable is created the current time is recorded which is then checked every 10 seconds. If after the time span either the primary memtable for the Column Family or any secondary index memtables have received writes they are replaced and flushed to disk.

Unfortunately as the wiki points out, there is a good reason to make this value small and a good reason to make it large.

A log file cannot be deleted until all of the segments / records it contains have been marked as completed. This happens when a memtable is flushed to disk. Because the log file is shared by all Column Familes, one Column Family that has long living memtables can prevent log files from been deleted and their disk space freed up.

However smaller values can cause multiple memtables to expire at the same time, prior to version 0.7 this could cause flush requests to block. Version 0.7 added memtable_flush_writers and memtable_flush_queue_size ,but it can still slow down the IO system (see conf/cassandra.yaml for more info). The best approach is to tune the other memtable thresholds to trigger when you want them, and leave this setting as a backup.

The wiki recommends setting a default of 1440 minutes, or 24 hours which is also the default.

memtable_operations (millions)

The Memtable tracks the number of operations applied to it by:

counting the number of top level columns (i.e. Columns for a Standard CF or Super Columns for a Super CF) in a mutation.
considering a row level deletion as a single operation.

Deletions and Insertions are both considered a mutation, inserting 3 columns increases the count by 3 and deleting 3 named columns increases the count by 3. Note that the number of sub columns in a Super Column is ignored. So inserting 10 sub columns into 1 super column increases the count by one, and deleting a super column by name that has 10 sub column increases the count by one.

The operations threshold (and the size threshold) is checked before applying a mutation and the flush is not requested until after the mutation has completed. So it’s possible for the memtable to contain more than memtable_operations when it is flushed to disk.

A different way to think about this setting, and memtable_throughput, is as sstable_min_operations and sstable_min_bytes. In general operation new sstables are created after at least sstable_min_operations operations have occurred or at most sstable_min_bytes bytes will be written.

If no value is provided when the CF is created it is set to the default memtable throughput in MB (below) / 64 * 0.3, so it’s 300k ops per 64MB of throughput. If you have a CF that contains many small columns it’s a good idea to look at the log entries for memtable flushes to see if the ops threshold is triggering early and causing small memtables to be frequently written.

memtable_throughput

Throughput for the memtable is tracked and tested at the same time as the operation count. But counting the byte size of the data is more involved and depends on the type of the column. The size of the data when serialised to disk is counted as follows:

standard column byte size is
- length of the key byte array plus 2 bytes to store the length
- 1 byte to indicate if the column has been deleted
- 8 bytes for the timestamp
- length of the value byte array plus 4 bytes to store the length
expiring columns (those with a TTL) add another 8 bytes to the length of a standard column.
deleted columns (tombstones) are the same as standard columns but the value is always 4 bytes long.
counter columns (in Cassandra v0.8) add another 8 bytes to the length of a standard column. Note that for a counter column the value will always be an 8 byte long.
super columns sum the size of all contained columns and then add
- length of the name byte array plus 2 bytes to store the length
- 4 bytes to indicate when it was deleted
- 8 bytes to store the timestamp for the deletion
- 4 bytes to store the number of sub columns

(Currently the calculation for the super column only includes the sum of the sub columns. I think this needs to be changed.)

A row deletion will add zero bytes to the throughput counter.

The byte size of the mutation is always added to the counter, if one mutation replaces columns in the memtable their byte size is not subtracted from the counter.

Getting this setting wrong is a very easy way to run out of memory. From version 0.7 onwards the worse case scenario is up to CF Count + Secondary Index Count + memtable_flush_queue_size (defaults to 4) + memtable_flush_writers (defaults to 1 per data directory) memtables in memory the JVM at once. It’s best to be conservative, follow the wiki advice and consider that the JVM may take up to 10 times as much memory as it takes to serialise the data to disk.

And that’s the problem with this threshold. It’s not measuring how much memory a memtable is using in the JVM Heap, it’s measuring the maximum amount of bytes it could take to serialise the data (excluding the index and bloom filter) to disk. Which makes it a difficult knob to use when tuning how much memory Cassandra uses.

If a value is not provided when the Column Family is created it will default to 1/16th the maximum size of the JVM Heap at the time. This value stored with the Column Family meta data and will not change again. Typical values are around 128MB to 256MB.

memtable_total_space_in_mb

Version 0.8 adds the per node memtable_total_space_in_mb setting which makes life easier and may eventually replace the 3 previous settings. While it’s fun to play with the per CF settings, it can also be a pain when building real systems that need to stay up.

If no value is set in conf/cassandra.yaml the setting will default to one third of the JVM max Heap size. If it is set to zero the setting is disabled and only the old per CF thresholds will be used. If the global setting is enabled and there are per CF settings both of them will be used.

There are two parts to the global memtable size, measuring the real memory usage of the memtable and flushing. First the measuring.

Rather than track every byte allocated the server periodically works out the ratio between the throughput as measured above and the real in memory bytes as measured by JVM. The in memory byte count is worked out using the Instrumentation Java Package and code from Jonathan Ellis. After a mutation has been applied to the memtable, but before a flush is requested, Cassandra calculates the “Live Ratio” if more than twice as many operations (as calculated above) have been processed since the last time it was calculated.

Measuring the Live Ratio is done asynchronously and involves measuring the real memory size of all the keys, super columns and columns in the memtable and dividing it by the throughput as measured above. For sanity the ratio is clamped between 1.0 and 64.0, if the value is outside of this range a WARN level log message will let you know. Finally the ratio for the Column Family is updated to the new ratio if and only if the new ratio is higher than the previous one. An INFO level message will let you know when the ratio is calculated, how long it took and if it changed.

Next the MeteredFlusher runs every second and uses a two phase approach to keeping the live memory use under the setting. First it looks at the total live bytes for each Column Family, including it’s secondary indexes, and flushes CF’s that could potentially fill the memory if allowed to create memtables of this size. Live bytes are calculated by multiplying the throughput as perviously measured by the Live Ratio. The Flusher considers the Column Family to be using too much memory if it’s current live size is more than memtable_total_space_in_mb divided by the maximum number of memtables the Column Family could have in memory. The calculation for this is similar to the one presented above for memtable_throughput but it includes secondary indexes and a fudge factor that takes into account how the live size is measured.

For example if memtable_total_space_in_mb is 100MB, and memtable_flush_writers is the default 1 (with one data directory), and memtable_flush_queue_size is the default 4, and a Column Family has no secondary indexes. The CF will not be allowed to get above one seventh of 100MB or 14MB, as if the CF filled the flush pipeline with 7 memtables of this size it would take 98MB. At a more sensible 2GB for memtable_total_space_in_mb (1/3 of a 6GB JVM Heap) the CF will be flushed if it is using 292MB of live memory.

(I’ve skipped a couple of things here such as considering the bytes currently been flushed.)

The flusher process will end there if the number of bytes that were flushing when it started plus the bytes for all the CF’s that were not flushed in the first phase is less than memtable_total_space_in_mb.

The second phase flushes the CF’s in order of largest to smallest until the total live size (including the bytes currently been flushed) gets down below the target setting.

This new setting (and the existing flush_largest_memtables_at) should make it harder to shot yourself in the foot with memory management and easier for new users to feel comfortable with the server.

In Motion

You can check the per CF thresholds as well as the current tracked values for a memtable using bin/nodetool, bin/cassandra-cli or JConsole. I’m not aware of any current features to check the Live Ratio or Live Size of a CF.

bin/nodetool cfstats can tell you the current operation count (‘Memtable Columns Count’) and throughput (‘Memtable Data Size’):

$ ./bin/nodetool -h localhost cfstats
Keyspace: dev
    Read Count: 1
    Read Latency: 0.897 ms.
    Write Count: 2
    Write Latency: 0.051 ms.
    Pending Tasks: 0
        Column Family: data
        SSTable count: 2
        Space used (live): 9530
        Space used (total): 9530
        Memtable Columns Count: 1
        Memtable Data Size: 26
        Memtable Switch Count: 1
        Read Count: 1
        Read Latency: 0.897 ms.
        Write Count: 2
        Write Latency: 0.020 ms.
        Pending Tasks: 0
        Key cache capacity: 200000
        Key cache size: 2
        Key cache hit rate: 0.0
        Row cache: disabled
        Compacted row minimum size: 51
        Compacted row maximum size: 86
        Compacted row mean size: 73

‘bin/cassandra-cli’ can tell you the current thresholds using either describe keyspace or show keyspaces.

[default@dev] describe keyspace;
Keyspace: dev:
  Replication Strategy: org.apache.cassandra.locator.NetworkTopologyStrategy
    Options: [datacenter1:2]
  Column Families:
    ColumnFamily: data
      Key Validation Class: org.apache.cassandra.db.marshal.BytesType
      Default column value validator: org.apache.cassandra.db.marshal.BytesType
      Columns sorted by: org.apache.cassandra.db.marshal.AsciiType
      Row cache size / save period in seconds: 0.0/0
      Key cache size / save period in seconds: 200000.0/14400
      Memtable thresholds: 0.29062499999999997/62/1440 (millions of ops/MB/minutes)

Or using JConsole connect to the server, select MBeans and then navigate to org.apacge.cassandra.db.ColumnFamilies.<your-keyspace>.<your-column-family>. There you can find the current thresholds:

MemtableFlushAfterMins
MemtableOperationsInMillions
MemtableThroughputInMB

And the running values:

MemtableColumnCount
MemtableDataSize

The forces of Write and Read

2011-04-28T00:00:00-07:00

Requests that write and read data in Cassandra, like any data base, have competing characteristics that need to be balanced. This post compares the approach taken by Cassandra to traditional Relation Database Systems.

When writing data the most efficient approach to splat it somewhere on disk with the minimum of fuss. Preferably at the end of a file, or even in a new file. The write is going to take longer if the data is carefully placed in the correct ordered position in an existing file.

Problem is reading data is going to be much more efficient if the data is in the correct ordered location, and preferably in just one location. If the data is unordered and spread out in a file, or multiple files, the read is going to take longer.

Traditional Relational databases such as SQL Server are optimised for read requests, so write requests ensure that data is written to the correct ordered location. However the transaction log / write ahead log lets the database delay writing the data to the correct location on disk, providing a handy performance boost while still supporting ACID transactions.

Roughly speaking Write requests

(very) Roughly speaking a write request in a RDBMS follows this path:

Append the write request to the end of the transaction log.
Locate the rows modified by the request or those adjacent to new rows, by reading index, data or other pages from disk if not already in memory.
Modify the index and data pages in memory to insert or update rows.
Acknowledge the write to the client.
Checkpoint the modified pages by flushing them to disk in the correct ordered location, and mark the log records associated with the changes as no longer needed. The Checkpoint process is typically executed asynchronously (SQL Server MySQL).

(I think in reality the log record is written once a statement completes inserting/updating an individual row. Also a record needs to be written to say if the transaction was committed or aborted, so that during log file recovery the system knows if the transaction should be replayed or ignored. I’m also ignoring all the locking.)

The write ahead log lets the write request dump the data somewhere on disk fast, then work on structures which are hopefully in memory. For updates care is taken to check for existing data, which is then read and mutated in memory. The read request ignores the transaction log, it reads the single source of truth from the correctly ordered index and data pages that are hopefully still in memory.

If the RDBMS uses Multi Version Concurrency Control life is a bit more difficult for the read request, so lets ignore that for now.

On the other hand Cassandra is optimised for write requests, so read requests need to do more work to ensure they have the correct data.

Again roughly speaking a write request for Cassandra follows the path:

Appended the write to the end of the write ahead log.
Modify an in memory structure called a Memtable to store the row changes present in the request.
Queue the Memtable to be flushed to disk (in the background) if per Memtable thresholds on operation count, data size or time to be violated. A new empty Memtable is immediately put in it’s place.
Acknowledge the write to the client.
Flush the Memtable to disk by creating a new SSTable (Sorted String Table) on disk, and mark the log records as no longer needed. Typically the flush task is executed asynchronously.

(This is the write process on a single node, I’m ignoring how the write is applied to the cluster and other features required for ‘Eventual Consistency’.)

Again the write ahead log lets the write process dump the data somewhere on disk fast, and then work on structures which are guaranteed to be in memory. No effort is taken to read the existing data, if a row was updated it may now partially exist in memory and in multiple SSTables on disk.

It’s probably important to point out that the RDBMS Checkpoint process works with an extent which defines it’s on disk allocation unit. Sql Server uses extents that contain 8 8KB pages. MySQL uses a default EXTENT_SIZE of 1MB for the extent, and the InnoDB engine has a hard coded 16KB page size. Once one extent has been written it may be necessary to seek to anther random part of the disk to write to next extent. Though I’m sure they do all sorts of clever things to make it nice and fast, this is the basic unit they deal with.

In Cassandra the maximum Memtable size is configured per Column Family (sort of like a table but I dislike that analogy). Memory thresholds of 128MB, 256MB or higher are common, though Memtables may flush at lower sizes due violating other thresholds or in response to certain system events. The Flush task is able to write the entire contents of the Memtable to disk as a new file without having to perform random IO seeks.

Dear Readers

In Cassandra read requests potentially have a lot more to do that those in a traditional RDBMS. A typical read request in a RDBMS to get a row by Primary Key seeks through the nicely ordered b-tree index pages to find the right data page, scans the page to find the appropriate row and reads the complete row (ignoring off page large data). Reading the index and data pages may require disk access, and a query cache may provide direct access to the results of a previous query without needing to negotiate the index.

Cassandra has two caches that can expedite a read request, a key cache (discussed below) and a row cache. The row cache contains all the current data for the row, and a query that needs to read any column from the row can be fully resolved by reading the cache.

If the requested row cannot be served from the row cache it must be read from disk, however the write path may have left fragments of the row in various SSTables. The read request follows the path:

Check the in memory Bloom Filter for each SSTable on disk, to build a list of candidate SSTables where the row may exist.
If enabled probe the key cache for each candidate SSTable from step 1 to get the position of the row in the data file. The key cache may hold only a sample of keys from the file so misses are possible.
If the key cache missed, probe the SSTable index summary held in memory which contains a regular sampling of the keys (and their positions) in the SSTable index file to find the preceding (sampled) key. Seek to the position of the preceding sampled key in the SSTable index file, and then scan the index file until the requested key and it’s data file position is found.
Seek to the row position in the SSTable data file and then scan the row data to read the columns that match the read request.
Reduce (potentially) multiple values for each column provided by each SSTable and the current Memtable to a single value by considering column timestamps, Time To Live settings and deletes.
Return the result to the client.

(This is the read path on a single node, I’m ignoring how the read is processed by the cluster, features required for ‘Eventual Consistency’, and have not mentioned that memory mapped file access is used by default on 64 bit platforms.)

The read request must piece together the row from potentially many files, which may each involve random IO. Each SSTable which contains any columns for a row must be read to determine if it contains columns that match the request criteria. Hot rows may be present in present in memory in the form of Cassandra key and row caches or OS caches.

Rows which receive many writes (including overwriting the same column) over time have the potential to perform worse than those that are written to within a short time span. Multiple writes to the same row and column are reconciled in memory by the Memtable so that only the current values for a row are written to disk in the SSTable. The Compaction Process provides an opposing force to the propensity of write requests to spread data out over multiple files.

With all this going on it’s still possible to get millisecond or better read performance. The import thing to note is that it’s possible to have poorly performing reads due to spread out rows.

Compacting for a sustainable future

The Compaction Process is there to help the read requests by reducing the number of SSTables they may have visit. Compaction is also responsible for finalising Distributed Deletes and Time To Live, but I will ignore those for now.

Major Compactions are triggered manually via the nodetool utility and compact all SSTables for a Column Family into one file. Owing to the way compaction chooses which files to process this may result in the new file not been compacted for a very long time. As a result Major compactions are no longer recommended as Minor compactions can do the same things. I’ve mentioned them here just for completeness.

Cassandra checks to see if a Minor compaction is needed whenever an SSTable is written to disk. The process groups the files into buckets where every file in a bucket is within 50% of the average size of files in the bucket, small files (less than 50MB) are put in the first bucket. If the bucket contains more than than the Column Family defined min_compaction_threshold files the compaction process will compact up to max_compaction_threshold files together.

During compaction the row fragments from the SSTables are reconciled to create a single row. When merging columns the one with the highest timestamp is used, if the timestamps are equal the column value is used as a tie breaker.

The result is a single SSTable that contains an aggregated view of the row that was present in the mulitple input SStables. And a shorter path for any read request that previously needed to potentially read from several SSTables.

In motion

It’s reasonably easy to see this process happening in slow motion by constructing a contrived schema in Cassandra and playing with some of the command line tools.

This schema creates two Column Families. The OneOp CF sets the Memtable to flush to disk after just one operation is written, and FiveOp after 5 operations (memtable_operations is expressed in millions of operations). The min_compaction_threshold tells compaction to start after we see 4 files. The script then inserts some sample data, all against the same row and the same columns to demonstrate overwrites.

Note that the Memtable thresholds are checked before the operation starts, but it is not flushed to disk until after the operation completes. For writes it counts the each columns in the request as an operation. In our case it means the SSTables will be created with after 2 or 6 operations rather than the expected 1 or 5.

Start the cassandra cli and paste the script into the cli.

create keyspace ReadWrite;

use ReadWrite;

create column family OneOp 
  with memtable_operations = 0.000001
  and min_compaction_threshold = 4;
create column family FiveOp 
  with memtable_operations = 0.000005
  and min_compaction_threshold = 4;

set OneOp['foo1']['bar'] = 'baz1';
set OneOp['foo1']['bar'] = 'baz2';
set OneOp['foo1']['bar'] = 'baz3';
set OneOp['foo1']['bar'] = 'baz3';
set OneOp['foo1']['bar'] = 'baz1';
set OneOp['foo1']['bar'] = 'baz1';


set FiveOp['foo1']['bar'] = 'baz';
set FiveOp['foo1']['bar'] = 'baz';
set FiveOp['foo1']['bar'] = 'baz';
set FiveOp['foo1']['bar'] = 'baz';
set FiveOp['foo1']['bar'] = 'baz';
set FiveOp['foo1']['bar'] = 'baz';
set FiveOp['foo1']['bar'] = 'baz';
set FiveOp['foo1']['bar'] = 'baz';
set FiveOp['foo1']['bar'] = 'baz';
set FiveOp['foo1']['bar'] = 'baz';
set FiveOp['foo1']['bar'] = 'baz';
set FiveOp['foo1']['bar'] = 'baz';

The data directory for the ReadWrite keyspace should now look like.

/var/lib/cassandra/data/ReadWrite:
total 200
drwxr-xr-x  22 aaron  wheel   748B 27 Apr 10:42 .
drwxr-xr-x   4 aaron  wheel   136B 27 Apr 10:42 ..
-rw-r--r--   1 aaron  wheel    75B 27 Apr 10:42 FiveOp-f-1-Data.db
-rw-r--r--   1 aaron  wheel    16B 27 Apr 10:42 FiveOp-f-1-Filter.db
-rw-r--r--   1 aaron  wheel    14B 27 Apr 10:42 FiveOp-f-1-Index.db
-rw-r--r--   1 aaron  wheel   4.2K 27 Apr 10:42 FiveOp-f-1-Statistics.db
-rw-r--r--   1 aaron  wheel    75B 27 Apr 10:42 FiveOp-f-2-Data.db
-rw-r--r--   1 aaron  wheel    16B 27 Apr 10:42 FiveOp-f-2-Filter.db
-rw-r--r--   1 aaron  wheel    14B 27 Apr 10:42 FiveOp-f-2-Index.db
-rw-r--r--   1 aaron  wheel   4.2K 27 Apr 10:42 FiveOp-f-2-Statistics.db
-rw-r--r--   1 aaron  wheel    76B 27 Apr 10:42 OneOp-f-1-Data.db
-rw-r--r--   1 aaron  wheel    16B 27 Apr 10:42 OneOp-f-1-Filter.db
-rw-r--r--   1 aaron  wheel    14B 27 Apr 10:42 OneOp-f-1-Index.db
-rw-r--r--   1 aaron  wheel   4.2K 27 Apr 10:42 OneOp-f-1-Statistics.db
-rw-r--r--   1 aaron  wheel    76B 27 Apr 10:42 OneOp-f-2-Data.db
-rw-r--r--   1 aaron  wheel    16B 27 Apr 10:42 OneOp-f-2-Filter.db
-rw-r--r--   1 aaron  wheel    14B 27 Apr 10:42 OneOp-f-2-Index.db
-rw-r--r--   1 aaron  wheel   4.2K 27 Apr 10:42 OneOp-f-2-Statistics.db
-rw-r--r--   1 aaron  wheel    76B 27 Apr 10:42 OneOp-f-3-Data.db
-rw-r--r--   1 aaron  wheel    16B 27 Apr 10:42 OneOp-f-3-Filter.db
-rw-r--r--   1 aaron  wheel    14B 27 Apr 10:42 OneOp-f-3-Index.db
-rw-r--r--   1 aaron  wheel   4.2K 27 Apr 10:42 OneOp-f-3-Statistics.db

There are 3 SSTables for OneOp and 2 for FiveOp, and all of them contain the a fragment of the ‘foo1’ row.

The nodetool cfhistograms utility shows recent statistics for requests against a Column Family, where recent means “since the last time it was run”. The same statistics are also available via the org.apache.cassandra.db.ColumnFamilies.<keyspace>.<column_family> MBean in JConsole.

Clear the stats for both Column Families on the command line.

$ bin/nodetool -h localhost cfhistograms ReadWrite OneOp
$ bin/nodetool -h localhost cfhistograms ReadWrite FiveOp

Now execute two reads via the cli.

[default@ReadWrite] get OneOp['foo1'];
=> (column=626172, value=62617a31, timestamp=1303875486224000)
Returned 1 results.
[default@ReadWrite] get FiveOp['foo1'];
=> (column=626172, value=62617a, timestamp=1303875486248000)
Returned 1 results.
[default@ReadWrite]

Check the stats for the OneOp Column Family.

$ bin/nodetool -h localhost cfhistograms ReadWrite OneOp
ReadWrite/OneOp histograms
Offset      SSTables     Write Latency      Read Latency          Row Size      Column Count
1                  0                 0                 0                 0                 3
2                  0                 0                 0                 0                 0
3                  1                 0                 0                 0                 0
4                  0                 0                 0                 0                 0

The histogram is saying that 1 request used 3 SSTables, and there are an estimated 3 columns in all the SSTables the Column Family is tracking. The Column Family should be tracking 3 SSTables, and each should have only one column for the ‘foo1’ row. The insert script overwrote the same column and the Memtable absorbed one overwrite before been flushed to disk.

The stats for the FiveOp Column Family also make sense.

$ bin/nodetool -h localhost cfhistograms ReadWrite FiveOp
ReadWrite/FiveOp histograms
Offset      SSTables     Write Latency      Read Latency          Row Size      Column Count
1                  0                 0                 0                 0                 2
2                  1                 0                 0                 0                 0

There are only 2 SSTables for the Column Family, both contain a fragment of the ‘foo1’ row and both contain only one column.

A minor compaction on the OneOp Column Family can be triggered by adding just two more wafer thin rows via the cli.

[default@ReadWrite] set OneOp['foo1']['bar'] = 'baz1';
Value inserted.
[default@ReadWrite] set OneOp['foo1']['bar'] = 'baz2';
Value inserted.
[default@ReadWrite]

This should trigger a minor compaction as there is now 4 SSTables of similar size. The data directory should now look something like this.

$ ls -l /var/lib/cassandra/data/ReadWrite/
total 280
-rw-r--r--  1 aaron  wheel    75 27 Apr 15:38 FiveOp-f-1-Data.db
-rw-r--r--  1 aaron  wheel    16 27 Apr 15:38 FiveOp-f-1-Filter.db
-rw-r--r--  1 aaron  wheel    14 27 Apr 15:38 FiveOp-f-1-Index.db
-rw-r--r--  1 aaron  wheel  4264 27 Apr 15:38 FiveOp-f-1-Statistics.db
-rw-r--r--  1 aaron  wheel    75 27 Apr 15:38 FiveOp-f-2-Data.db
-rw-r--r--  1 aaron  wheel    16 27 Apr 15:38 FiveOp-f-2-Filter.db
-rw-r--r--  1 aaron  wheel    14 27 Apr 15:38 FiveOp-f-2-Index.db
-rw-r--r--  1 aaron  wheel  4264 27 Apr 15:38 FiveOp-f-2-Statistics.db
-rw-r--r--  1 aaron  wheel     0 27 Apr 16:16 OneOp-f-1-Compacted
-rw-r--r--  1 aaron  wheel    76 27 Apr 15:38 OneOp-f-1-Data.db
-rw-r--r--  1 aaron  wheel    16 27 Apr 15:38 OneOp-f-1-Filter.db
-rw-r--r--  1 aaron  wheel    14 27 Apr 15:38 OneOp-f-1-Index.db
-rw-r--r--  1 aaron  wheel  4264 27 Apr 15:38 OneOp-f-1-Statistics.db
-rw-r--r--  1 aaron  wheel     0 27 Apr 16:16 OneOp-f-2-Compacted
-rw-r--r--  1 aaron  wheel    76 27 Apr 15:38 OneOp-f-2-Data.db
-rw-r--r--  1 aaron  wheel    16 27 Apr 15:38 OneOp-f-2-Filter.db
-rw-r--r--  1 aaron  wheel    14 27 Apr 15:38 OneOp-f-2-Index.db
-rw-r--r--  1 aaron  wheel  4264 27 Apr 15:38 OneOp-f-2-Statistics.db
-rw-r--r--  1 aaron  wheel     0 27 Apr 16:16 OneOp-f-3-Compacted
-rw-r--r--  1 aaron  wheel    76 27 Apr 15:38 OneOp-f-3-Data.db
-rw-r--r--  1 aaron  wheel    16 27 Apr 15:38 OneOp-f-3-Filter.db
-rw-r--r--  1 aaron  wheel    14 27 Apr 15:38 OneOp-f-3-Index.db
-rw-r--r--  1 aaron  wheel  4264 27 Apr 15:38 OneOp-f-3-Statistics.db
-rw-r--r--  1 aaron  wheel     0 27 Apr 16:16 OneOp-f-4-Compacted
-rw-r--r--  1 aaron  wheel    76 27 Apr 16:16 OneOp-f-4-Data.db
-rw-r--r--  1 aaron  wheel    16 27 Apr 16:16 OneOp-f-4-Filter.db
-rw-r--r--  1 aaron  wheel    14 27 Apr 16:16 OneOp-f-4-Index.db
-rw-r--r--  1 aaron  wheel  4264 27 Apr 16:16 OneOp-f-4-Statistics.db
-rw-r--r--  1 aaron  wheel    76 27 Apr 16:16 OneOp-f-5-Data.db
-rw-r--r--  1 aaron  wheel  1936 27 Apr 16:16 OneOp-f-5-Filter.db
-rw-r--r--  1 aaron  wheel    14 27 Apr 16:16 OneOp-f-5-Index.db
-rw-r--r--  1 aaron  wheel  4264 27 Apr 16:16 OneOp-f-5-Statistics.db

The SSTables for OneOp numbered 1 through 4 have been compacted, this is tracked in the server and a compacted marker file (e.g. “OneOp-f-4-Compacted”) is written to disk to preserve the information across system restarts. The unused files will be physically deleted during JVM Garbage Collection. If Cassandra detects it is low on disk space when about to write data to disk, it will trigger GC in an effort to reclaim unused space.

There is also a new SSTable OneOp-f-5-Data.db that contains the single reconciled row from the other 4 SSTables. A read request against that row should now only use one SSTable:

[default@ReadWrite] get OneOp['foo1'];
=> (column=626172, value=62617a32, timestamp=1303877811057000)
Returned 1 results.
[default@ReadWrite]

Check the stats.

$ bin/nodetool -h localhost cfhistograms ReadWrite OneOp
ReadWrite/OneOp histograms
Offset      SSTables     Write Latency      Read Latency          Row Size      Column Count
1                  1                 0                 0                 0                 1
2                  0                 0                 0                 0                 0
3                  0                 0                 0                 0                 0
4                  0                 0                 0                 0                 0

The request used one SSTable and the Column Family has only 1 column in all the SSTables it is tracking.

Introduction to Cassandra

2011-02-07T00:00:00-08:00

I’ve tidied up a previous Introduction to Cassandra presentation with better diagrams and improved explanations. It’s available as Keynote, PDF or plain old web pages. It’s a basic introduction to the way Cassandra works as a clustered system. It covers Partitioning, Replication, Hinted Handoff, Read Repair and Anti Entrophy. It does not cover the data model.

Presentation classic is still here

If you have any questions or suggestions let me know. You can also use it as a starting point if you are creating a Cassandra presentation of your own.

Hello World

2011-01-31T00:00:00-08:00

How are you?

Aaron Morton

PRIMARY KEY's in CQL

Schema, Schema, Every Where

The Setup

The Sound of One Column Indexing

Partitioning and Clustering

The Composite Enchilada

Sorting Lists For Humans

The Natural Order

Take It and Turn It

Ledger List

What’s Missing ?

Row Level Isolation and Consensus

We Can All Agree on Consensus

Assign23

CF23

The Fine Print

ZooKeeper Reading 12-01-2012

Get a connection

Get the children of a znode

Get the properties of a znode

Tree operations

znode operations

Watch for changes

Cassandra Reading 02-01-2012

Anatomy of a Cassandra Partition

Is Cassandra available ?

Orientate

Stop and think

Generational issues

Back to the future

Back to the present

Step 1 - Snapshot

Step 1 - Delete the Generation column

Step 3 - Snapshot

Step 4 - Check the delete

Step 5 - Compact away the Tombstone.

Step 6 - Confirm the purge

Step 7 - Re-write the Generation number

Step 8 - Confirm the Generation re-write

Step 9 - Restart node #23

Wrapping up

Reversed Comparators

In Motion - The Setup

In Motion - Start Column

In Motion - Count Only

Token Range Ownership

Cassandra Query Plans

Lots-o-garbage

What’s in a row?

Slice By Names

Slice From Read

So What ?

Name Locality

Start Position

In Motion - The Setup

In Motion - Name Locality

In Motion - Start Position

Down For Me?

Which nodes ?

The weird Consistency Level

Consistency Levels

A failing Range

A failing Cluster

The view from one node

Recent Cassandra AWS Reading

User List discussions

Blogs and such

Deletes and Tombstones

Remembering things forgotten

Local Reads for Local Queries

Readers Digest

Read Repair (sort of)

Free

The reincarnation of Paul Revere’s horse

In Motion on a single node

In Motion on a cluster

How are Memtables measured?

memtable_flush_after (minutes)

memtable_operations (millions)

Get the children of a `znode`

Get the properties of a `znode`

`znode` operations