Secondary Indexes Design

Table of Contents

Index key fields for a table

• Index fields (and other necessary information, like say the type of index) to be declared at the level of the table. Indexes can be added or dropped any time.
• Internally, key indexed field is assigned an enumerated value.
• Later: Allow indexes on multi-keys. i.e., can have an index which is just one key, or two or more keys.

Index keys for objects

Now, RAMCloud knows what fields to build keys on. How do we figure out these keys for each object? RAMCloud doesn't understand the object blob, the application does.

 Solution: Provide index keys explicitly, apart from the object (key-value).

• write(tableId, key, value, ..., numKeys, Array<indexName, indexLength>)

Note that the first entry in the array must correspond to the primary key.

What would the actual stored object look like?

• key0 is primary key, always and has to be present.
• The offsets are the relative positions of the keys. By definition, they are cumulative.
• Offset 0 points to beginning of key 0.
• Using offsets makes key length computation trivial.
• The key offsets and keys are put in the order of index enumeration.
  • For i >= 1, if key_i is not present, then keyOff_i = keyOff_i-1. This basically means the length of the i^th key is 0.
• If the application is smart, it could not include data from keys in the value, and use it directly from the index fields when returned.

Index location

If index and table can both fit on one server, great!

But what if they don't?

Options:

Index on one or more index servers (these index servers could have data too, but the point is that they don't have to). Partitioned based on index key: 

• Index distributed based on index key on various index servers (these index servers could also have other data).
• Coordinator keeps a map of which parts of which secondary indexes reside on which servers
• Lookup analysis below - (1) - This is option (a)
• Recovery times for data table server and index table server will be different. :-/
• TODO: What if the list of keys (to multiread on) returned from index lookup are too long to fit in one rpc?

• Any index lookup would go to all servers on which index is located.
• Data (for each table) is scattered across servers and get re-scattered during recovery. Re-distributing index according to data would be an additional overhead during recovery.
• Lookup analysis below - (1) - This is option (b)
• This will certainly be "better" for writes (no rpcs to index writes, hence lower latency) and better for reconstruction recovery.

<index key, object pointer> (as in primary key hash table)

• Not possible if index entries not on the same server as corresponding data records. Also, during if the object is moved (for example, during recovery), then index entries would have to be updated.

<index key, primary key>

• Avoids problems with previous option.
• Primary key could be large, at least larger than obj ptr.

<index key, primary key hash> 

• Better than 2 since hash table is based on primary key hash anyway, so why not store the hash directly – saves time and space.
• Since primary key hash may not uniquely identify object, need to compare read object with the key provided to ensure we have the right object(s).
• Concern: How to do garbage collection?
  • Problem: If an object A is deleted, we look at the secondary index. When we are looking at a secondary index entry with a 64 bit primary key hash X, how do we know that there wasn't another object B with the same 64 bit primary key hash X that had the same secondary index value (and hence this entry corresponds to another object, and hence can't be deleted). Sure, the likelihood of this is very very slim, but we have to consider it.
  • Solution idea: Keep a count in the secondary index entries for each primary key hash corresponding to this index key. Doesn't work since incrementing / decrementing this count is not an idempotent action – and we need all our operations to be idempotent (so that transactions will work).
  • Solution: Lookup hash X in the primary hash table, then compare the entire key hash (>64 bit) where .....

Index format

Initial thoughts below. More up to date info to be maintained by Ashish on a different wiki page.

Tree: Enables range queries apart from lookups.

• Modular (tree implementation should be changeable without affecting rest of the implementation)
• API assumptions: Allows lookup, add, remove. All ops are atomic and idempotent.
• Our "distribution" strategy is to partition based on index key. Many distributed tree implementations do random node placement, which will be highly inefficient / not scalable.
• Tree structure: Check related work (B+ Tree, Adaptive Radix tree, MassTree, BWTree, etc) and figure out which one(s) would work best. If possible, make modular so that it can be changed if needed.
• Concurrency issues? Is it good enough to take a global lock on the tree partition?
• Later: Trie for string key indexes and normal tree for numerical key indexes? Might help with space efficiency. Would be more complex since need two implementations.

Hash table: Enables lookups on exact match. Options:

• Make it just like the current primary key hash table. Good: Can reuse current implementation. Bad: This index would have to be modified every time data moves. (See section on index entry format.)
• Index key points to the primary key, then another lookup in primary key hash table to get the object itself. 
  • Will need a mechanism to handle duplicates. (Since secondary keys don't have to be unique, could have multiple keys here.)

To begin, could implement trees, since that's where we get the max functionality boost. Later want to implement both and provide option while defining indexes.

Initial thoughts below. More up to date info to be maintained by Ashish on a different wiki page.

There will be some index (for some part of some table) and some data (from the same or another table) on a given server. So index trees will have to share memory with segment storage for data. How do we allocate (and clean) memory to the index? Options:

• Stuff index nodes into RAMCloud key-value objects.
  • Might be inefficient – TODO analysis.
  • Simpler since these can get managed (allocated / read / cleaned / recovered) in the same way as regular RAMCloud objects.
  • If we want to use backup/recover option for recovery, then this is particularly good.
• Allocate log segments to index and use these segment spaces to store parts of the tree.
  • How will this allocation and cleaning work?
  • This approach might be convenient if we can use RAMCloud recovery for recovering the index too.
• Split memory between index and data.
  • Dynamic splitting not possible without changing RAMCloud memory allocation (and don't want to go there).
  • Static splitting will require information a priori about index and data size which might not be available, else result in underutilized space.
• Completely separate index servers and data servers.
  • Then there would be issues about how many servers to allocate for indexes and how many for data?
  • How would this impact index server recoveries?
  • Memory allocation can then be done independently for data and index without interference from the other.

Consistency

Definition

At any point, data is consistent with index entries corresponding to it:

Required:

1. If data X exists, X is reachable from all key indexes.
2. Data returned to client is consistent with key used to look it up.

If these are not satisfied, then linearizability is violated. But there's a tradeoff with performance. In future, consider providing an option to turn this on or off.

For multi-threaded read/write non-locking access, no failures

(Can directly look at Object Update, since that includes the other cases.)

Simplest solution for all cases, wrap steps into a transaction. But, we'd like to avoid transactions as far as possible (to retain as much performance and scalability as possible).

Various cases:

Object(s) lookup based on secondary index:

• Lookup in the index, return keyhashes corresponding to query
  • If index should exist (based on index declaration) but doesn't exist right now, then maybe its being reconstructed after a crash.
    • Handling options: keep retrying till success  or return "retry" to client 
• Client does multiread on these keyhashes
  • For any keyhash, if no match, then return empty result for that object (just as in normal read)
  • If matches found for a keyhash, check we have the right objects based on index keys. Return matching values.

Object write: Options:

1. Insert index entry then write object:  -> Object available (via all indexes) as soon as object written.
   1. Read before either insertion / write: no inconsistency
   2. Read after both insertion and write: no inconsistency
   3. Read between insertion and write: index entry found, matching object not found, read will return no value, as if object wasn't written yet. Same thing with multiple indexes.
2. Write object then insert index entry:
   1. Read before either write / insertion: no inconsistency
   2. Read after both write and insertion: no inconsistency
   3. Read between write and insertion: index entry not found, read will return no value, as if write hadn't completed. This is inconsistent since object does exist. If we have multiple indexes, then lookup on one could succeed and lookup on another could fail.

Object delete: Options:

1. Remove index entry then delete object
   1. Violates property 1.
   2. Example of undesirable situation: If I have index A and index B on same data, then removing entry from index A then from B then actually deleting data could cause a situation where lookup on index A doesn't return value, but lookup on index B returns a value.
2. Delete object then remove entry:  -> Object not available (via any index) as soon as object deleted.
   1. Read on this object before deletion and removal: object read, no inconsistency
   2. Read after deletion and removal: object not found, no inconsistency
   3. Read between deletion and removal: index entry found, corresponding object not found, no object returned.
   4. Note: delete() operation doesn't have information about indexes. Hence, before / while deleting object, read in the index keys so that they can be removed. How to handle failures?

Object update:

From the consistency requirement and RAMCloud operation linearizability, it follows that we want the following linearizability property for object updates:

There exists time x, s.t. before x, client can lookup old data; at and after time x, it can lookup the new data.

Have to remove old entry AND insert new entry. Combination of object delete and object write.

Key idea: Writing object is the commit point.

Example: Primary key is "Foo" which hashes to 4444444. Its value is initially Bob Brown and later updated to Sam Smith. There are indexes on the first name (Bob / Sam) and last name (Brown / Smith). Timeline:

Note: The old and new index entries may not be on the same physical server.

For multi-threaded write/write non-locking access, no failures

Previous state: Foo: Bob Brown. Client 1 writes: Foo: Sam Smith. Now, say client 2 writes: Foo: Jen Jones.

Allowed end states: Consistent data and index entries:

Data - Foo: Sam Smith; fname - Sam: 444444; lname - Smith: 444444 ---OR--- vice versa

Not allowed end states: Data X but index entries corresponding to Y:

Data - Foo: Sam Smith; fname - Jen: 444444; lname - Jones: 444444 ---OR--- vice versa

Not ideal end states: Consistent data and index entries plus dangling pointers:

Data - Foo: Sam Smith; fname -  Sam: 444444, Jen: 444444; lname - Smith: 444444, Jones: 444444 ---OR--- vice versa

It seems like there is no interleaving of actions leading to not allowed state.

Potentially bad interleaving produces ideal result:

1. Master writes index entries for SS
2. Master writes index entries for JJ
3. Master reads BB and writes data SS (this invalidates BB)
4. Master reads SS and writes data JJ (this invalidates SS)
5. Master removes index entries for BB
6. Master removes index entries for SS

Potential Solution : (If a write involves indexes,) serialize writes to the same object including index updates. Lock update on object before step 1 and release after step step 3. – Implications: This makes batching of index deletes not feasible. Also, cannot push the indexing logic to client anymore.

Failure of an Index Server

The client request being processed gets blocked till the index server is recovered; completed and return to client once index server recovers.

In the meanwhile, two things can happen:

1. The client might retry this request
   • The write (including index updates) is idempotent by design, so not a problem.
2. Another client might send another request for the same object
   • This is the same as multi-threaded access to the same object problem.

If an update involves this index server, return RETRY to client. Can leave dangling pointers around. Failure (and returning retry) at any point will not violate required consistency properties.

No need to return RETRY if failure happens at removal of index entries stage.

Implementation-wise: Use RPC wrapper to throw error if index server failed.

For a reasonable first implementation: Just wait for index server to recover, no need to return RETRY.

Failure of Master Server

Consider the sequence of steps for an update: insert new index entry, update object, remove old index entry. Lets say they happen on the same server (i.e., index and data in this case located on the same server). What if the server fails at some point here?

Simple solution: Wrap all the ops into a transaction. But as in previous subsection, we'd like to avoid transactions as far as possible. So, for now, lets assume no transactions.

Failure scenarios:

Failure before the update begins or after it is complete are okay – ok – don't affect anything

Failure after step 1, during step 2 – Case A:

• Current state maintains consistency properties stated above
• There are dangling pointers (to new unwritten data)
• On client retry:
  • Step 1 redone: Idempotent, so not a problem
  • Step 2-3 done: Client operation completes
• If the client dies and never retries?
  • If index restoration is done by rebuilding, then eventually, when this server crashes/recovers, the dangling pointers will go away.
  • If index restoration is done by backup/recovery, then these may stay around forever. 

Cleaning

An index cleaner process can scan indexes in background and delete index entries that are dangling pointers.

Complication: By design, we have dangling pointers for a little (write index entries, then update / write the object). Will have to detect this and not delete such index entries.

Another concern: Since indexes and data may not be on the same server, cleaning on an index could require an rpc for each object in the index. Can batch with multiread which will make it much more efficient.

Approach (that addresses the above concerns too): Steps:

1. Index cleaner starts running on an index partition (index partition = part of one index corresponding to one table that is located on one server).
2. Send (new) rpc confirmExists(.., .., .., ..) with all the index entries as params that gets bucketed and sent to all the masters owning data from this rpc (similar to multiop mechanism).
3. Each master server checks if that exists (or is being currently written), and for data that doesn't exist, send (new) rpc multiRemoveFromIndex(..., ...) to the index server.