This looks promising. Is there such thing as a generalized SQL query engine that runs over any key-value store that provides certain minimal core operations?
For example, say you have a KV Store with basic mathematical Set operations like GET, SET, UNION, INTERSECT, EXCEPT, etc. The Engine would parse the SQL and then call the low-level KV Store Set operations, returning the result or updating KV pairs. This explains how Join relates to Set operations:
Another thing I'd like is if KV stores exposed a general purpose functional programming language (maybe a LISP or a minimal stack-based language like PostScript) for running the same SQL Set operations without ugly syntax. I don't know the exact name for this. But if we had that, then we could build our own distributed databases, similar to Firebase but with a SQL interface as well, from KV stores like Pulsar. I'm thinking something similar to RethinkDB but with a more distinct/open separation of layers.
The hard part would be around transactions and row locking. A slightly related question is if anyone has ever made a lock-free KV store with Set operations using something like atomic compare-and-swap (CAS) operations. There might be a way to leave requests "open" until the CAS has been fully committed. Not sure if this applies to ledger/log based databases since the transaction might already be deterministic as long as the servers have exact copies of the same query log.
Edit: I wrote this thinking of something like Redis, but maybe Pulsar is only the message component and not a store. So the layering might look like: [Pulsar][KV Store (like Redis)][minimal Set operations][SQL query engine].
Pulsar isn't a KV store though, it's a distributed log/messaging system that supports a "key" for each message that can be used when scanning or compacting a stream. GET and SET aren't individual operations but rather scans through a stream or publishing a new message.
If you just want to have a SQL interface to KV stores or messaging systems that support a message key then Apache Calcite [4] can be used as a query parser and planner. There are examples of it being used for Kafka [5].
One of the challenges with layering SQL on top of a KV store is query performance.
The most obvious way to model a secondary index on top of a pure KV store is to map indexed values to keys. For example, given the (rowID, name) tuples (123, "Bob"), (345, "Jane"), (234, "Zack"), you can store these as keys:
name:Bob:123
name:Jane:345
name:Zack:234
At this point you don't need or even want values, so this is effectively a sorted set.
Now you can easily find the rowID of Jane by doing a key scan for "name:Jane:", which should be efficient in a KV store that supports key range scans. You can do prefix searches this way ("name:Jane" finds all keys starting with "Jane"), as well as ordinal constraints ("age > 32", which requires that the age index is encoded to something like:
age:Bob:\x00\x00\x00\x20:123
To perform an union ("name = 'Bob' OR name = 'Jane'"), you simply do multiple range scans, performing a merge sort-ish union operation as you go. To perform an intersection ("name = 'Bob' AND age > 10"), you find the starting point for all the terms and use that as the key range, then do the merge sort.
This is what TiDB and FoundationDB's record layers do, which both have a strict separation between the stateless database layer and the stateful KV layer.
The performance bottleneck will be the network layer. Your range scan operations will be streaming a lot of data from the KV store to the SQL layer, and potentially you'll be reading a lot of data that is discarded by higher-level query layers. This is why TiKV has "co-processor" logic in the KV store that knows how to do things like filter; when TiDB plays your query, it pushes some query operators down to TiKV itself for performance.
Unfortunately, this is not possible with FoundationDB. This is why FoundationDB's authors recommend you co-locate FDB with your application on the same machine. But since FDB key ranges are distributed, there's no way to actually bring the query code close to the data (as far as I know!).
I'm sure you could do something similiar with Redis and Lua scripting, i.e. building query operators as Lua scripts that worked on sorted sets. I wouldn't trust Redis as a primary data store, but it can be a fast secondary index.
FoundationDB's client bindings have a locality API which allows you to query the client's metadata cache of which key ranges are on which storage processes. This would allow you to build that feature of routing a query to the data.
I disagree that an index is a set rather than a map: it is by definition a map from keys to row ids.
As for a generic relational layer over K/V stores, I think it’s a superficially appealing idea that would be impossible to optimize adequately in practice. Honestly, if you want to implement a distributed relational database, I would recommend starting with Postgres as your local storage engine and pushing down as much relational logic as possible to that layer. I have worked on such a system in the past and it produced very good results for minimal development effort.
I'm not talking about just any index, I'm talking about using a plain KV store as an index.
The row ID can be encoded into the key. From my example, a basic mapping might be:
name:Bob:123 => <empty value>
If your KV store is optimized for range scans, as they usually are, then there's no reason to store anything in the value, because a key range scan can efficiently jump to the first instance of a key prefix.
For example, if I want to search for "name = 'Bob'", then I simply start at the key "name:Bob:" and pluck the row ID from each key, scanning sequentially until I reach the end of my range.
This works great for multiple values. For example:
Finding all rows matching "Bob" is a matter of just scanning by prefix.
If you store row IDs in the value, you'll risk read/write contention on the value. Let's say there are multiple rows with "Bob", you'll end up storing something like:
name:Bob => [123, 124, 125]
To add or remove row IDs you'll now have to merge values of unrelated rows, and make sure this happens atomically. That usually means locking.
This also puts a constraint on the number of row IDs you can fit in a single value. KV stores typically co-locate value data with keys, so now you might not be able to efficiently scan a large range without also loading that data. You can do tricks like introducing an indirection, where you don't store "row IDs", but "row page IDs", where each page is sharded maximum of N row IDs, allowing you to sidestep the size limit on values. But that comes with other costs.
I'm not counting in-memory stores like Redis here. Implementing in-memory indexes is a completely different ball game to something that needs to live on disk.
As for "superficially appealing idea that would be impossible to optimize adequately in practice", your assertion is demonstrably false: TiDB and CockroachDB both implement performant relational databases on top of general-purpose KV stores.
e.g. how about a complex event processing engine? Something like that will do a lot of the above, but the inference database stays managable since old data will fall out of the windows.
For example, say you have a KV Store with basic mathematical Set operations like GET, SET, UNION, INTERSECT, EXCEPT, etc. The Engine would parse the SQL and then call the low-level KV Store Set operations, returning the result or updating KV pairs. This explains how Join relates to Set operations:
https://blog.jooq.org/2015/10/06/you-probably-dont-use-sql-i...
Another thing I'd like is if KV stores exposed a general purpose functional programming language (maybe a LISP or a minimal stack-based language like PostScript) for running the same SQL Set operations without ugly syntax. I don't know the exact name for this. But if we had that, then we could build our own distributed databases, similar to Firebase but with a SQL interface as well, from KV stores like Pulsar. I'm thinking something similar to RethinkDB but with a more distinct/open separation of layers.
The hard part would be around transactions and row locking. A slightly related question is if anyone has ever made a lock-free KV store with Set operations using something like atomic compare-and-swap (CAS) operations. There might be a way to leave requests "open" until the CAS has been fully committed. Not sure if this applies to ledger/log based databases since the transaction might already be deterministic as long as the servers have exact copies of the same query log.
Edit: I wrote this thinking of something like Redis, but maybe Pulsar is only the message component and not a store. So the layering might look like: [Pulsar][KV Store (like Redis)][minimal Set operations][SQL query engine].