Encoded Data in Columnstores

What Is Encoded Data and What Does It Mean to Operate on It?

SingleStore supports several different kinds of data encodings to help compress data. All columnstore data is said to be encoded in one way or another. Common encodings are run-length, dictionary, and LZ4. Only certain encodings can be processed directly, i.e. operated on directly. These include dictionary encoding, run-length encoding, and integer value encoding. For example, with dictionary encoding, for a segment of a string column with only 3 distinct values, a 2-bit ID number is stored for each string. The ID numbers are used as references into the dictionary. These IDs are packed tightly together when stored in the columnstore.

For example, consider this dictionary:

This can be represented as a bit vector of packed string IDs when it is stored into the column segment:

• Strings: red, blue, green, green, red

• String IDs: 0, 2, 1, 1, 0

• Bit vector: 00 10 01 01 00 (2 bits per string ID)

As an example of operating directly on encoded data, SingleStore can perform a filter operation, say t.a = xyz, on a string dictionary-encoded column segment by first finding the result of the filter for every entry in the dictionary. Then, while scanning the segment, the query execution system simply takes the encoded ID number for each value in t.a and uses it to look up the result of the string comparison for that ID that was computed in the initial scan of the dictionary. That tends to be much quicker than doing an actual string comparison. Other kinds of operations can also be done directly on encoded data ID values, including the comparison of ID values needed to do group-by operations. The details of these are beyond the scope of this documentation.

The encoding for a blob depends on its type and data. It may hinge both on the ordering and cardinality for numeric delta encoding and dictionary encoding, respectively. It impacts the execution speed, as well as enables SIMD operations on encoded data.

Most query processing in SingleStore outside of columnstore scan is done row-at-a-time. Columnstore processing on encoded data is done in a vectorized fashion, where large batches of data from a column are processed in one or more relatively simple loops. These loops tend to be more friendly to modern CPUs than row-at-a-time processing, resulting in lower instruction count, better cache usage, and improved efficiency of the processor’s instruction pipeline.

Queries That Can Benefit from Operations on Encoded Data

Queries that can benefit from operations on encoded data are those that run over columnstore tables, and have one or more of these components:

• Filters

• Group-by

• Aggregates

• Expressions in aggregates

• Expressions in group-by

• Star joins on integer columns

Operations on encoded data can be done regardless of the number of distinct values of the encoded columns being processed. Performance tends to be better when the number of distinct values is smaller. This is because when dictionaries are smaller the columnstore data is compressed to a smaller size, and intermediate working lookup tables fit more easily into the cache memory of the processor.

Here’s an example of a query that can benefit from operations on encoded data in several ways.

SELECT
  DAYOFWEEK(shipdate), location,
  SUM(price * (1 - discount)), AVG(quantity)
FROM lineitem
WHERE
  status = 'S'
GROUP BY 1, 2;

Here’s another example of a query that can benefit from join operations on encoded data, if the table f is a columnstore.

SELECT d1.x, d2.y, SUM(f.m)
FROM f, d1, d2
WHERE f.d1_key = d1.key
  AND f.d2_key = d2.key
  AND d1.v > 100
  AND d2.v in (1, 2, 3)
GROUP BY d1.x, d2.y

This query is a simple example of a star join. If you are using a star schema (with a fact table linked to multiple dimension tables), and the joins are on integer columns, you can benefit from operations on encoded data. Use of a group by clause and aggregate functions is required to benefit from join operations on encoded data.

Examples

The following example shows encoded group-by in action:

CREATE DATABASE db;
USE db;
DROP TABLE IF EXISTS r;
DROP TABLE IF EXISTS is_tables;

/* Extract rows for 50 different tables into a scratch table, is_tables. */
CREATE TABLE is_tables AS
SELECT * FROM information_schema.tables
LIMIT 50;

/* Create a table with a million rows, with every table name
   appearing in most or all segments. */
CREATE TABLE r (INDEX USING CLUSTERED COLUMNSTORE(n)) AS
SELECT s.*, (row_number () OVER (ORDER BY s.table_name) % 1000) AS n
FROM (SELECT t1.*
      FROM is_tables t1, is_tables t2, is_tables t3, is_tables t4
      LIMIT 1000000) AS s;

The table r created above is a columnstore with one million rows. You can see the encoding of the TABLE_NAME column using this query:

SELECT COUNT(*), encoding, column_name
FROM information_schema.columnar_segments
WHERE table_name  = 'r' AND database_name  = 'db'
  AND column_name = 'TABLE_NAME'
GROUP BY column_name, encoding;

The result indicates that StringDictionary encoding is used for the column.

Now, run the following group-by/aggregate query that groups on the table_name column, in profile mode:

PROFILE SELECT table_name, COUNT(*) FROM r GROUP BY table_name;

Finally, output the JSON profile for the query:

SHOW PROFILE json;

The result includes the following text:

inputs":[
    {
        "executor":"ColumnStoreScan",
        "db":"db",
        "table":"r",
        ...
        "segments_scanned":{ "value":16, "avg":2.000000, "stddev":0.000000, "max":2, "maxPartition":0 },
        "segments_skipped":{ "value":0 },
        "segments_fully_contained":{ "value":0 },
        "segments_encoded_group_by":{ "value":16, "avg":2.000000, "stddev":0.000000, "max":2, "maxPartition":0 },
        "inputs":[]
    }

Notice the portion labeled segments_encoded_group_by. This part shows that 16 segments were processed in total and all of them were processed with encoded group-by operations. These operations are taking place directly on encoded data.

Now, here is another query (run in profile mode) that is similar to the first, but also has a filter on column table_name:

PROFILE SELECT table_name, COUNT(*) FROM r
WHERE table_name LIKE '%COLUMN%'
GROUP BY table_name;

SHOW PROFILE json;

The JSON profile for this query contains the text below:

"inputs":[
    {
        "executor":"ColumnStoreScan",
        "db":"db",
        "table":"r",
        ...
        "segments_scanned":{ "value":16, "avg":2.000000, "stddev":0.000000, "max":2, "maxPartition":0 },
        "segments_skipped":{ "value":0 },
        "segments_fully_contained":{ "value":0 },
        "segments_ops_compressed_data":{ "value":16, "avg":2.000000, "stddev":0.000000, "max":2, "maxPartition":0 },
        "segments_encoded_group_by":{ "value":16, "avg":2.000000, "stddev":0.000000, "max":2, "maxPartition":0 },
        "inputs":[]
    }

Notice that it includes sections for both segments_ops_compressed_data and segments_encoded_group_by. The segments_ops_compressed_data section indicates that 16 segments had filters applied on encoded data.

The final decision about use of operations on encoded data is made at runtime and depends on column encodings for each segment and (for group-by) the number of distinct values in the segment. So the segments_ops_compressed_data and segments_encoded_group_by values may be less than the number of segments scanned minus the number skipped. In this case, the encoded_group_by_bailout section will indicate the reasons that encoded group by was disabled as well as the number of segments disabled for each reason.

/* Create a simulated dimension table with 1000 rows with unique
   values for n, and around 100 categories. */
CREATE TABLE d AS
SELECT n, MAX(CONCAT('cat', n % 100)) AS category
FROM r
GROUP BY n;

/* Join dimension table d to "fact" table r on an integer column
   and group by category. This is a very simple star join. */
SELECT d.category, COUNT(*)
FROM r, d
WHERE r.n = d.n
  AND d.category LIKE 'cat1%'
GROUP BY d.category;

SELECT a, COUNT(*) FROM t GROUP BY a;

Performance Considerations

Because operations on encoded data can process data so fast, their performance can become limited by the bandwidth of main memory (RAM). And the bandwidth of RAM is normally far higher than the bandwidth of your I/O system. For example, RAM bandwidth could be 50 GB/sec and the bandwidth of an SSD could be only 600 MB/sec.

So, SingleStore recommends that you configure your system so that little or no I/O is being done while processing columnstore data, under a steady query workload. You can do this by ensuring that the operating system’s file system buffer cache has enough main memory to hold the working set of your columnstore data, i.e. the segments of columns that are frequently accessed. A good rule of thumb is to have enough RAM in the file system buffer cache to hold at least 20% of your compressed columnstore data. If your workload does a lot of full table scans that touch all the columns of the table, you may want to consider adding more than that.

The Linux operating system will allocate available system memory to the file system buffer cache in an on-demand way, without the need to set any configuration knobs. Simply ensure that there is sufficient memory remaining after all the memory dedicated to SingleStore, the operating system, and any other applications running on the hardware.

Relevant Hints

SingleStore can perform group-by on encoded data using a HashGroupBy operation, and can also do a group-by on columnstore data using a StreamingGroupBy operation on the sort key of the columnstore. Due to the improvement of HashGroupBy with the introduction of operations on encoded data, it may be the case that HashGroupBy is faster than StreamingGroupBy, yet the query optimizer may choose StreamingGroupBy. If this occurs and it is important for your workload to tune the query to get better performance, you can use the disable_ordered_scan query hint. For example:

SELECT SUM(a) FROM t WITH (disable_ordered_scan=true) GROUP BY b;

You can use EXPLAIN to see the plan for your query to check the type of group-by operation chosen.

Data Encodings Supported

Operations on encoded data are supported only for the following:

• String columns (including varchar, char, text, and blob) with string dictionary and string run-length encodings

• Numeric columns (including all integer types (tinyint, bigint, etc.) plus float, decimal, and double) with value and run-length encodings

Group-by/aggregate operations on encoded data are supported only for integers. Filter acceleration on encoded data is supported for both strings and integers.

SingleStore recommends that you allow SingleStore to choose automatically how to encode data for columnstores. However, in the rare event that it chooses an encoding that does not support operations on encoded data, and it is important to you for your application that operations on encoded data be performed, you may wish to force the encoding chosen. You can do this with the option '<encoding>' notation. For example:

CREATE TABLE t (a int, b varchar(50) option 'StringDictionary',
  key(a) USING CLUSTERED COLUMNSTORE);

Operations Supported, and Limitations

To summarize the operations supported on encoded data, as well as limitations to operations on encoded data, they include:

• Scan

  • Faster decoding for integer encoding

• Filter

  • Filters on the following:

    • Strings for string dictionary and string run-length encoding

    • Integers for run-length encoding

  • Or-of-ands of filter expressions; filter expressions must involve a single string column for faster processing

  • Support for Bloom filters on string columns (elimination of rows with no matches for joins on a single string column)

• Aggregates

  • Aggregates supported: sum, min, max, count, any

  • Aggregate input data types supported: all numeric

  • Aggregate expressions supported: multiple table columns allowed in a single expression

  • Aggregates encodings supported: integer, integer run-length

• Group-by:

  • Operations on encoded data are not done for aggregates without a group-by (a.k.a. scalar aggregates)

  • Group-by count star is very fast

  • Group-by:

    • Multiple group-by columns are allowed

    • A mix of columns and expressions is allowed

    • Group-by columns must only use these encodings: integer, integer run-length

    • There is a limit on number of distinct values per column of a few thousand rows, beyond which, the system reverts to row-at-a-time processing and the local aggregation of data for a row segment will output rows to the parent global aggregate operator

    • Similarly, there is a limit on total number of groups, of a few thousand groups, beyond which the optimized group-by processing is not used

    • Group-by on the sort key may not be as efficient as group-by on other columns because ordered group-by may be performed while hash groupby might be better

  • Group-by on expressions:

    • Only one input column is allowed in a single expression

    • The expression result must be integer

    • Each table column can be used only once in the set of group-by columns and expressions

• Joins:

  • Joins must be on some type of integer column, or another column type represented internally as an integer, such as datetime

  • The join or joins selected by the query optimizer must be of type HashJoin

  • The join must be a many-to-one relationship

  • The HashJoin (or a sequence of HashJoins) must appear over a ColumnStoreScan

  • A Group By operation and aggregate must be present in the query over the result of the join

Encoded group by bailout reasons

The final decision on whether to use encoded group by is made at runtime. The possible bailout reasons are:

• Join condition is not a many-to-one relationship

• Unsupported column encoding

• Aggregate not guaranteed to fit in aggregate type (the result may overflow the internal integer or decimal type)

• Reached dynamic dictionary size limit (too many unique group by values)

• Group by column cannot be placeholder blob (column does not yet exist in columnstore)

Additional Tuning Considerations

If you have upgraded from a MemSQL release below 6.0, you can benefit immediately from operations on encoded data. There is no need to rebuild your columnstore tables. However, columnstore encoding choices have changed slightly in the 6.0 release, with dictionary encoding being more preferred. So queries may run faster on newly loaded data. Encodings are chosen separately for each segment, so even if you don’t use OPTIMIZE on your columnstore tables, as new data is inserted and older data is removed, the encodings chosen will naturally evolve to be those preferred by the latest version of SingleStore.

If you find that operations on encoded data are not occurring as much as you expect them to you can rebuild the columnstore table involved. Before doing this, you can query information_schema.columnar_segments as described earlier to see if dictionary, run-length, or integer encodings are not used. If not, then consider rebuilding the columnstore table.

One way to rebuild it is to use

CREATE TABLE newTable (INDEX USING CLUSTERED COLUMNSTORE(<columns>)) AS
SELECT *
FROM oldTable;

Then drop oldTable and rename newTable to oldTable. Alternatively you can use OPTIMIZE TABLE… FULL. However, be aware that this can take significantly longer than loading the data in the first place, because it causes a single sorted run to be created.

SELECT a, COUNT(*) FROM t GROUP BY a;

SELECT COUNT(*) FROM t;

SELECT SUM(t2.c)
FROM (SELECT a, COUNT(*) AS c FROM t GROUP BY a) AS t2;