Note 0x03: Sharded & Materialized Columns
Improving query throughput when working with bulky, dynamic, semi-structured data.
Pattern Overview
When someone says “sharding”, we almost always think about tables, but what about individual columns?
If a column contains a relatively large payload (e.g., json, text, or map), but most of our queries need only a portion of this payload, one method of improving the query throughput is rather trivial yet often forgotten: split this column into multiple smaller columns.
When working with columnar databases the column shards can often be added to the same table. For row-oriented databases, however, the column shards usually have to be moved into new tables to minimize the amount of data read from disk, the process known as vertical sharding. Even in columnar databases, storing shards in a separate table might be desired if the process of backfilling historical data into the new table structure is computationally expensive or otherwise difficult.
In real-world scenarios, when running analytical workloads over sharded columns, we observe an average performance improvement of x2-10 times, proportional to the column size reduction. However, if we dedicate a shard to hold only a single key of the payload (i.e., introduce a materialized column), the performance can improve by over 100x due to better data locality and vectorized execution.
In practice, we often combine column sharding with individual column materialization. The column sharding can be introduced up-front without knowledge of schema. The keys of the payload can be distributed across shards using bucketing, consistent hashing, or any other technique of your choice. Column materialization, however, requires knowledge of the keys the payload contains. The keys can be inferred from the data, or from the observed query patterns. In general, introducing a dedicated column is a more expensive operation, as it involves schema migration and backfill, yet offers the best performance improvement.
Many systems defer column materialization to preserve flexibility of the storage system. The data is first ingested with no write-time contract enforcements, then certain fields get conditionally promoted into dedicated columns to improve query performance. The columns can be added/dropped dynamically. The ingestion pipeline does not need to adapt to schema changes.
Practical Applications
1. Tags/labels: We often use key-value maps such as “tags” in our data models. Sharding a map into N columns is a very inexpensive approach to improving query throughput. The following blog post mentions Netflix using this technique.
Instead of having a large “tags” column, we introduce “tags_00”, “tags_01”, … “tags_N” columns.
At ingestion, we split the tags map into N maps using a hash function over the tag key, that is, tags_00={k: v in tags.items() if hash(k) % N == 0}, tags_01={k: v in tags.items() if hash(k) % N == 1}, and so on.
At query time, when a user requests the dataset to be filtered by e.g. “... where tag[A] = B” we calculate the tag’s shard as “hash(A) % N”, then rewrite the query into “... where tag_<shard>[A] = B”.
2. Logging and event analytics: The benefits of schema-agnostic ingestion and on-demand column materialization are well described at this Uber blog post that goes over their logging platform.
The logs from services have a number of fields (tags) that engineers attach to track the context. These field names change over time. Moreover, the queries used to search for logs also change over time. This makes the approach of on-demand adaptive column materialization an attractive pattern.
In Uber’s case, the raw log payload is stored in a “_source” column. This column is sharded into type-specific maps: “_strings”, “_numbers”, “_booleans”, although there is still an option to further shard these maps into N sub-maps.
Frequently used fields are materialized on-demand into dedicated columns, e.g. “alter table add column v_status_code_number default _numbers[‘status_code’]”. The service layer maintains the awareness of materialized columns, and rewrites the queries to reference them instead of maps, e.g. use “v_status_code_number” instead of “_numbers[‘status_code’]”.
Infrequently used materialized columns are periodically dropped.
3. Schemaless: Schema-agnostic OLTP stores, such as key-value and their variations, offer two primary benefits that enable them to scale well with organizational growth: the elimination of invasive database schema migrations and the scalability derived from their enforced persistence model. Uber’s Schemaless, Pinterest’s Oversharded MySQL, core principles of FoundationDB’s Architecture, and other examples illustrate the benefits of such storage engines.
The technique is similar to what’s been discussed. The storage engine persists key-value pairs, where the value is a blob (e.g., json, protobuf). The schema is maintained on the application side. If there is a need for an index over the field stored in a blob, the field is materialized into a new key-value pair, by the application.
Hands-On Example
Note that some database engines already have native support for complex data types, such as maps and json, and might implicitly perform column sharding and materialization, as demonstrated in the following Clickhouse demo. The decision to offload that to the database, or keep it database-agnostic by keeping it on the application side, is fully yours.
Below is a sketch of the method using Clickhouse, launched in Docker:
The following script is executed within the Clickhouse shell:
Key Takeaways
Column sharding and materialization can be used to improve query performance when working with bulky, semi-structured data, like maps or json.
Column sharding involves splitting a large column into multiple smaller columns. Column materialization promotes a key from the complex payload into its own dedicated column. While sharding can be introduced without prior schema knowledge, materialization requires knowing the specific keys/fields to promote.
Many systems defer column materialization, allowing flexible, schema-agnostic data ingestion, followed by on-demand promotion of fields into separate columns.
The note is also provided as a PDF file, for convenience.
