Contents

• In-Memory Columnar Store (IMCS)
  • Introduction
  • Overview
  • Functions
    • General columnar store functions
    • Generated data manipulation functions
    • Generated data access functions for single timeseries
    • Generated data access functions for multiple timeseries (identified by timeseries ID)
    • Timeseries constructors
    • Binary operations
    • Unary operations
    • Mathematical functions
    • Date/time functions
    • Binary scalar functions
    • Timeseries transformation functions
    • Grand aggregates
    • Group-by aggregates
    • Grid aggregates
    • Window (moving) aggregates
    • Hash aggregates (group-by using hash function)
    • Cumulative aggregates
    • Sort functions
    • Special functions
  • Operators
  • Projection issues
  • Implementation details
  • Installation and tuning
  • Performance comparison
  • License

In-Memory Columnar Store (IMCS)

• Introduction
• Overview
• Functions
• General columnar store functions
• Generated data manipulation functions
• Generated data access functions for single timeseries
• Generated data access functions for multiple timeseries (identified by timeseries ID)
• Timeseries constructors
• Binary operations
• Unary operations
• Mathematical functions
• Date/time functions
• Binary scalar functions
• Timeseries transformation functions
• Grand aggregates
• Group-by aggregates
• Grid aggregates
• Window (moving) aggregates
• Hash aggregates (group-by using hash function)
• Cumulative aggregates
• Sort functions
• Special functions
• Operators
• Projection issues
• Implementation details
• Installation and tuning
• Performance comparison
• License

Introduction

Columnar store or vertical representation of data allows to achieve better performance in comparison with classical horizontal representation due to three factors:

1. Reducing size of fetched data: only columns involved in query are accessed.
2. Vector operations. Applying an operator to set of values (tile) makes it possible to minimize interpretation cost. Also SIMD instructions of modern processors accelerate execution of vector operations.
3. Compression of data. Certainly compression can also be used for all the records, but independent compression of each column can give much better results without significant extra CPU overhead. For example such simple compression algorithm like RLE (run-length-encoding) allows not only to reduce used space, but also minimize number of performed operations.

Overview

As it is clear from the abbreviation (IMCS: In-Memory Columnar Store) this plugin adds to PostgreSQL in-memory columnar store. So vertical representation of data is complementary to standard horizontal representation. Data is imported in PostgreSQL database in usual way and is stored in normal table. Then columns from this table are fetched and stored in shared memory. IMCS provides a lot of timeseries functions which can be used for data analysis. Operations with timeseries are performed in vector mode allowing to reach maximal possible speed of such operations. Also IMCS makes it possible to parallelize execution of some queries (for example calculation of aggregates) and utilize all CPU cores. All this three factors: in-memory location of data, vector operations, parallel query execution, makes it possible to increase speed of some queries more than 100 times comparing with standard PostgreSQL queries.

To make access to timeseries as convenient as possible, IMCS provides generator of access functions. You should specify name of source table or view (from which data will be imported), name of timestamp field (this is a main key by which timeseries elements are accessed) and optionally timeseries identifier. The last one needs some explanations. In some cases all data from the table should be placed in a single timeseries. For example assume that we collect data about phone calls (date, duration, caller, callee,...). It will be a single timeseries. But for example in trading systems there are separate data (ticks) associated with each symbol. So we have separate timeseries for ABB, GOOG, IBM, YHOO,... In this case securities identifier (symbol) can be considered as identifier of timeseries.

IMCS supports the following element type for timeseries: "char", int2, int4, date, int8, time, timestamp, float4, float8, bpchar. All timeseries elements should have the same size, so only fixed size character types are supported: for example char(10), but not varchar. Also IMCS is not able to represent NULL values. It is not enforced that fields of the source table were declared as NOT NULL, but attempt to insert NULL value in timeseries will cause error (or optionally NULL can be substituted with zero). Please use default values instead of NULLs.

Given all this information IMCS generates corresponding types and functions for loading/appending/accessing this timeseries. Assume that we have table Quote. After calling cs_create('Quote', 'Day', 'Symbol') we will get Quote_load() function for loading data from table in memory, Quote_get(symbol char(10), low date, high date) function for fetching/slicing corresponding timeseries and triggers which will keep track updates in Quote table and propagate this changes to timeseries.

There are two ways of synchronizing original table and timeseries:

1. Automatic: using triggers. In this case all inserts/deletes in original table are immediately reflected in timeseries.
2. Manual: using explicit invocation of load/append/delete methods.

When data is loaded from the table, records are sorted by timestamp and inserted in ascending order. You can append data to existed timeseries, but timestamps of inserted elements should be greater than already loaded. When timeseries is populated using insert trigger it is necessary to enforce that the data is inserted in the table in timestamp ascending order. Otherwise out-of-order error will be reported while inserting element in timeseries.

TABLE_get functions returns row of type TABLE_timeseries (this type is also generated by IMCS) which has the same columns as original table, but type of this columns is timeseries. So it is possible to refer to this timeseries as to any other columns and apply timeseries functions to them. For example query:

    select cs_max(Close) from Quote_get('IBM');

    select cs_avg(High - (Open + Close)/2) from Quote_get('IBM');

    select cs_filter(Open 

When you print result of execution of this query at the screen (for example by running this query in psql), it will be represented as large string literal in braces: 'date:{01/01/2010, 01/02/2010,...}'
Certainly it is not convenient for really large timeseries and may even cause memory exhaustion. 
Alternatively it is possible to change  vertical representation back to horizontal representation using TABLE_project or cs_project functions. Then produced tuples can be accessed in normal way using all SQL stuff.
For example it is possible to sort them or perform more grouping/filtering.
 



Functions
General columnar store functions

General columnar store functions are used to generate table-specific API functions, get information about columnar store and perform cleanup.

Generated data manipulation functions

Generated functions for loading/storing/deleting timeseries.

Generated data access functions for single timeseries

Functions generated for accessing single timeseries (timeseries having no identifier).

    select (Quote_project(abb.*,cs_top_max_pos(Close, 10))).* from Quote_get('ABB', date('01-Jan-2010'), date('31-Mar-2010')) abb;
    select (Quote_project(abb.*)).* from (select Symbol,Day,cs_maxof(Open,Close),High,Low,cs_minof(Open,Close),Volume from Quote_get('ABB')) abb;

    select (Quote_project(abb.*)).* from (select Symbol,cs_maxof(Open,Close) from Quote_get('ABB')) abb;

    select cs_project(abb.*) from (select Symbol,cs_maxof(Open,Close) from Quote_get('ABB')) abb;

    select p.* from (select Symbol,cs_maxof(Open,Close) from Quote_get('ABB')) abb, cs_project(abb.*) p(symbol char(10), max real);

Generated data access functions for multiple timeseries (identified by timeseries ID)

Functions generated for accessing multiple timeseries (source table contains identifier of timeseries, for example 'Symbol').

Timeseries constructors

Functions constructing constant timeseries (timeseries of repeated value) or timseries created by parsing string literal.

Binary operations

Binary operations with timeseries. These functions take two timeseries arguments and return result timeseries. IMCS tries to automatically adjust types of input arguments (for example if one timeseries has "int8" element type and another - "float8", then first one will be converted to "float8").

Unary operations

Unary operations with timeseries. These functions take single timeseries and return timeseries as result.

Mathematical functions

Calculation of mathematical functions for all timeseries elements.

Date/time functions

Extracts components of date/time type. These functions are mostly needed in group-by operations to calculate aggregates for various intervals (days, weeks, months, quarters, years...).

Binary scalar functions

Functions of this group take two timeseries arguments and calculate single scalar value as result. IMCS tries to automatically adjust types of input arguments (for example if one timeseries has "int8" element type and another - "float8", then first one will be converted to "float8").

Execution of these functions can be parallelized.

Timeseries transformation functions

Functions performing various transformations of input timeseries.

    select cs_call(Close, 'sin'::regproc) from Quote_get('IBM');
    select cs_call(Close, 'sin(float)'::regprocedure) from Quote_get('IBM');

    create function mul2(x real) returns real as $$ begin return x*2; end; $$ language plpgsql strict immutable;
    select cs_call(Close, 'mult2'::regproc) from Quote_get('IBM');

Grand aggregates

Functions calculating grand aggregates: aggregation of all timeseries elements.

Execution of these functions can be parallelized (except cs_median).

Group-by aggregates

Functions calculating aggregates for each group. Groups are identified by sequence of elements with the same value in group-by timeseries. It is not mandatory to sort this timeseries. But you should realize that sequences of the same value in different parts of the timeseries will form different groups. For example, there are four groups in timeseries '{1, 1, 2, 1, 1, 1, 2, 2,}': ('{1, 1}', '{2}', '{1, 1, 1}', '{2, 2}'). If you want to perform aggregation for all timeseries element with the same value, then use cs_hash_* functions instead.

Grid aggregates

Splitting timeseries into intervals intervaland calculating aggregate for each interval (grid cell).

Window (moving) aggregates

Aggregation is done for window - N subsequent elements of timeseries where N is window size. At each step window is moved at one position forward. So result timeseries has the same number of elements as input timeseries. First N-1 elements of result are calculated for windows smaller than N. You can use cs_limit(cs_window_AGG(input, N), N-1) to skip these elements.

Hash aggregates (group-by using hash function)

Aggregation with group-by. These function perform grouping and aggregation similar to SQL. All elements of group_by sequence with the same value forms single group. It is done using hash function, so cs_hash_* aggregates require additional memory for building hash table. These functions contain two out parameters: return two timeseries. The first one contains calculated aggregates. The second one contains corresponding group keys.

If it is necessary to perform grouping by more than one key, it is possible to use cs_cat (or || SQL operator) to concatenate several columns. Later it is possible to use cs_cut or cs_as functions to split concatenated value back into components.

Execution of these functions can be parallelized.

Cumulative aggregates

Aggregates calculated for all preceding elements of timeseries. Result timeseries has the same number of elements as input timeseries.

Sort functions

Top functions find out top-N values of timeseries. N can not be larger than imcs.tile_size (default value 128).

Execution of cs_top_* functions can be parallelized. Please notice that calculation of TOP-n is much faster than full sort.

Special functions

This group contains functions performing quite complex precessing of timeseries.

create type char16 as (body char(16));
select agg_val,cs_as(group_by,'char16') from (select (cs_project_agg(cs_hash_sum(volenquired,fxvenue))).* from DbItem_get()) agg;

select agg_val,encode(btrim(group_by, E'\\000'::bytea), 'escape') from (select (cs_project_agg(cs_hash_sum(volenquired,fxvenue))).* from DbItem_get()) agg;

select agg_val,cs_as(group_by,'char16')::char16 from (select (cs_project_agg(cs_hash_sum(volenquired,fxvenue))).* from DbItem_get()) agg;

create function to_char16(body bytea, type_name cstring default 'char16') returns char16 as '$libdir/imcs', 'cs_as' language C stable strict;
select agg_val,(to_char16(group_by)).* from (select (cs_project_agg(cs_hash_sum(volenquired,fxvenue))).* from DbItem_get()) agg;

Operators

IMCS provides standard SQL operators for timeseries type plus some specific operators for timeseries processing. The following table contains mapping between operators and corresponding timeseries functions:

Please notice that operators & | # in PostgreSQL have precedence different from precedence of standard SQL AND OR operators. Please always use parenthesis.

Binary operators accept not only timeseries OP timeseries operands. Also it is possible to pass as right parameter one of

1. Numeric constant (integer or floating point): select Price*2.0 from Quote_get()
2. Date, time or timestamp: select Day=date('11-Nov-2013') from Quote_get()
3. String literal (1): select Close*'{2.0,2.1,2.2}'::text from Quote_get('11-Nov-2013', '13-Nov-2013')
4. String literal (2): select Close*'float4:{2.0,2.1,2.2}' from Quote_get('11-Nov-2013', '13-Nov-2013')

In first two cases constant timeseries (timeseries containing the same value) is implicitly created for right operand using cs_const function. In third case timeseries is created from string literal using cs_parse function. And in the last case conversion to timeseries type is implicitly made by PostgreSQL using input function of this type.

Projection issues

There are several functions in IMCS API returning a row or a set of rows: cs_hash_*, cs_project*, cs_as. PostgreSQL provides two ways of decomposition of compound type into columns:

1. select (foo()).*
2. select * from foo()

    select (cs_project_agg(cs_hash_sum(Close,Day%7))).* from Quote_get('IBM');

First of all it is possible to avoid (...).* construction and access composite type attributes explicitly:

    select (q.p).agg_val,(q.p).group_by from (select cs_project_agg(cs_hash_sum(Close,Day % 7)) p from Quote_get('IBM')) q;

    select (q.p).agg_val,(q.p).group_by from (select cs_project_agg(cs_hash_sum(Close,Day % 7)) p from Quote_get('IBM') offset 0) q;

And concerning case 2) calling function in FROM list: it is possible if function doesn't depend on other data sources at the same query layer. For example in IMCS cs_hash_sum accepts timeseries arguments which are provided by Quote_get. So we can write:

    select (p).* from Quote_get('IBM') q,cs_project_agg(cs_hash_sum(q.Close,q.Day%7)) p;

Also using projection function in FROM list allows to specify alias and describe columns:

    select (s).* from Quote_get('IBM') q,cs_project(cs_hash_sum(q.Close,q.Day%7)) s(sum float8,group_by integer);

Conclusion:

1. Avoid using of (cs_hash_AGG(...)).* construction to eliminate redundant calculations.
2. Better use IMCS with PostgreSQL 9.3 and build queries using lateral joins.

Implementation details

Timeseries are stored in shared memory as B-Tree pages. This B-Tree provides fast access to timeseries element by position (for all types) or by value (only for timestamp). There is separate B-Tree for each timeseries. PostgreSQL hash is used to locate timeseries by identifier. Hash key includes name of the source table, name of the corresponding field and optionally identifier of timeseries. For example for Quote table identifier of timeseries may be 'quote-close-IBM'. Size of B-Tree pages is determined by "imcs.page_size" configuration parameter. Default value is 4kb.

IMCS uses RW (read-write) lock to synchronize access to columnar store. It means that multiple read-only queries can be performed concurrently, but adding or removing timeseries elements is possible only in exclusive mode. Lock is set when timeseries is accessed first time. If imcs.serializable configuration parameter is true (default), then lock is hold till the end of transaction. Such locking policy provides serializable isolation level for timeseries. If imcs.serializable is false, then lock is released at the end of query execution. It corresponds to "read committed" isolation level.

Right now IMCS doesn't provide any compression of timeseries elements. It can be done in the future. But IMCS is first of all oriented on financial data (trading systems). And here duplicates are not so often. Even if value of some stock option is quite durable (variation about few cents per year), small fluctuations of this option during a days normally occur. Our experiments show that RLE encoding cause only degrade of performance in case of standard queries on securities data.

Most of cs_* functions are not actually performing any executions. Instead of it, they are constructing pipe of operators (or expression tree). A unit of exchanging data between operators is a tile (vector). So operators are performing vector operations to reduce interpretation cost. Size of a tile should be large enough to minimize overhead of organizing work of the pipe. But it should fit in L1 CPU cache to keep processing speed high. Default size of the tile is 128.

IMCS is able to execute some operations in parallel. Now it is done for grand and hash aggregates, top-N functions (all operators where size of output is smaller than size of input). IMCS maintains pool of threads. Number of threads in the pool can be specified using "imcs.n_threads" configuration parameter. By default (zero value of this parameters), number of threads is detected automatically based on number of CPUs (cores) in the system. IMCS clones expression subtree and splits into segments timeseries accessed in the leave nodes of this tree (timeseries stored in columnar store). Number of segments corresponds to number of threads. So each thread is processing its own part of timeseries. Then results are merged using operator-specific merge function. Merge requires synchronization, so only one thread can perform merge at each moment of time.

Please notice that PostgreSQL is not able to parallelize execution of SQL query. Certainly it is possible to manually split query into several subqueries and execute them concurrently. But it is not trivial and not convenient. The fact that IMCS can overcome this limitation is very important for OLAP queries.

Installation and tuning

As far as IMCS is using PostgreSQL shared memory, it should be loaded via shared_preload_libraries. Please add '$libdir/imcs' to shared_preload_libraries in postgresql.conf file:

    shared_preload_libraries = '$libdir/imcs'		# (change requires restart)

    #define PG_MMAP_FLAGS			(MAP_HUGETLB|MAP_SHARED|MAP_ANONYMOUS|MAP_HASSEMAPHORE)

Below is list of all IMCS configuration parameters:

So steps of using IMCS extension are the following:

1. Change PostgreSQL configuration file postgresql.conf by adding IMCS to list of preloaded libraries and specifying maximal size of IMCS storage.
2. Install IMCS extension (read PostgreSQL documentation about installation of extensions).
3. Create extension using create extension imcs command (you need to have superuser permissions for it).
4. Generate interface functions using cs_create function.
5. If data is already present in the database, load it in columnar store using TABLE_load() function.
6. You should either enable autoload (imcs.autoload configuration property), either manually call TABLE_load() each time you restart the server.

Performance comparison

Consider the following definition of Quote table:

create table Quote (Symbol char(10), Day date, Open real, High real, Low real, Close real, Volume integer);

\copy Quote from 'NYSE_2003-2013.csv' with csv header;

create extension imcs;
select cs_create('Quote', 'Day', 'Symbol');
select Quote_load(); 

Now let's calculate volume-weighted price for IBM for the period from 2010 till 2013:

select cs_wavg(Volume, Close) as VWAP from Quote_get('IBM', date('01-Jan-2010'), date('01-Jan-2013'));

select sum(Close*Volume)/sum(Volume) from Quote where Symbol='IBM' and Day between date('01-Jan-2010') and date('01-Jan-2013');

Now let calculate VWAP for all symbols. To simplify it we will first create table containing information about all symbols. Actually this data is usually available and contains much more information than just symbol name. But here we need just symbol name:

create table Securities (Symbol char(10));
insert into Securities select distinct Symbol from Quote;
create view SecurityQuotes as select (Quote_get(Symbol)).* from Securities;

select Symbol,cs_sum(Close*Volume) / cs_sum(Volume) as VWAP from SecurityQuotes;

select Symbol,sum(Close*Volume)/sum(Volume) as VWAP from Quote group by Symbol;

Now let's test filter queries with projection back to horizontal representation. The following query finds all dates for 'ABB' symbol when close price was more than 1% large than open price for the particular quarter:

select (Quote_project(abb.*,cs_filter_pos(Close>Open*1.01))).* from Quote_get('ABB', '01-Jan-2010', '31-Mar-2010') abb;

select * from Quote where Symbol='ABB' and Day between date('01-Jan-2010') and date('31-Mar-2010') and Close>Open*1.01;

Actually timeseries in this example are not long enough: size of timeseries for each symbol is about 2608 elements. Let's now investigate use case with single large timeseries:

select Quote_drop();
select cs_create('Quote', 'Day');
select Quote_load();

select Volume//Close as VWAP from Quote_get();

select sum(Close*Volume)/sum(Volume) as VWAP from Quote;

Now perform filter query for large timeseries:

select cs_count((Close>Open*1.1)?) from Quote_get();
select count(*) from Quote where Close>Open*1.1;

Now consider real use case wth one timeseries and large enough records. There are about 10 million records with ~40 columns. Database size is about 5Gb. Queries perform groupping by various combinations of fields and cacluate aggregates for some characterestics. For example:

select trader,desk,office,sum(score*volenquired)/sum(volenquired) from DbItem group by trader,desk,office;

select agg_val,cs_cut(group_by,'c22c30c10') from 
    (select (cs_project_agg(ss1.*)).* from 
          (select (s1).sum/(s2).sum,(s1).groups from DbItem_get() q, 
               cs_hash_sum(q.score*q.volenquired, q.trader||q.desk||q.office) s1,  
                cs_hash_sum(q.volenquired, q.trader||q.desk||q.office) s2) ss1) ss2;

So summarizing these results: IMCS provides about 5-10 times increase of performance for relatively small timeseries (thousands elements) and 100 times faster speed for large timeseries (millions elements) on standard desktop with quad core processor. For SMP server with larger number of cores this ratio is expected to be even higher.

License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the Software), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHOR OF THIS SOFTWARE BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

Please send any feedbacks, complaints, bug reports, change requests to the author Konstantin Knizhnik. Latest version of this software can be obtained from the site http://www.garret.ru.