Partitioning Data in kdb+

select from trade where int=hour 2020.06.27D16 
select from trade where int within hour 2020.06.26D0 2020.06.27D16

hour:{`int$sum 24 1*@[;0;-;1970.01.01] `date`hh$\:x} 
intToDate:{1970.01.01+x div 24}

q)hour 2020.06.27D16 
442576i 
q)intToDate 442576i 
2020.06.27

\d .u 
n:("J"$.z.x 2)*`long$1024 xexp 2;

if[n<=hcount L;endofpart[]]

p:{
 f:x where x like (get `..src),"_*";
 $[count f;max "J"$.[;((::);1)]"_" vs'string f;0]
 } key `:.

.u.addLookup:{
 `:lookup/ upsert .Q.en[`:.] raze {select part:enlist x,tab:enlist y,
 minTS:min time,maxTS:max time from y}[x] each tables[]
 };

k)saveAndReload:{[h;d;p;f] 
(@[`.;;0#].Q.dpft[d;p;f]@)'t@>(#.:)'t:.q.tables`.; 
if[h:@[hopen;h;0];
 h"system"\l .\";cacheLookup[]";>h]
 };

q)quote:([]time:`timestamp$();sym:`symbol$();bid:`float$();ask:`float$();bsize:`int$();asize:`int$())
q)trade:([]time:`timestamp$();sym:`symbol$();price:`float$();size:`int$())
q)upd:insert
q)-11!`sym_0
12664
q)div[;1024] sum calcSize each (trade;quote)
q)4204

The resulting estimate is 4204KB. Comparing this to the size of the same data as stored on disk (uncompressed) results in a similar 4244KB:

$du -s HDB/0
4244    HDB/0

The main flaw with the calcSize function is its inability to calculate the size of data in array columns, such as the string type. It could be extended to account for this but then it’s complexity and run time would increase as it would need to integrate each cell rather than using only the first row as it does in its basic form.

Kdb+ itself provides a shortcut to calculate the IPC serialized size of an object with -22!:

q)div[;1024] sum -22!/:(trade;quote)
3710

While optimized for speed -22! remains an expensive operation. It also gives inaccurate results for symbol type data as in memory they are interned for efficiency but during IPC transfer use varying space depending on their length.

In common with calcSize both these methods also suffer from being unable to account for the memory overheads associated with any columns which have attributes applied to them.

In the process itself .Q.w can be interrogated to view actual memory reserved in the heap and used by objects:

q)div[;1024] .Q.w[]`heap`used
65536 4702

Whilst .Q.w in the RDB may seem like a good way to trigger in practice having the tickerplant poll another process is not a good idea as it is designed to be a self-contained process which will reliably store the transaction log and never be able to be blocked a downstream process, it publishes data asynchronously for this reason.

Overall this is an area where the “keep it simple, stupid” principle applies. There is little benefit to attempting to be too exact. Choosing a simple method and allowing a cautious RAM overhead for any inaccuracy is the best path to follow.

Handling late data

Filter time buffer

The hour helper is exact, this may be to exact for some use cases. For example a table with multiple timestamp columns which are created as the data flows through various processes. These other timestamp columns will be slightly behind the final timestamp created in the tickerplant.

Note: This issue is not limited to int partitioning and can be beneficial in any partitioned database.

If a user queries without accounting for this they could be presented with incomplete results:

select from trade where 
 int within hour 2020.06.26D0 2020.06.26D07, 
 otherTimeCol within 2020.06.26D0 2020.06.26D07

This can be manually accounted for by adding a buffer to the end value of your time window. Here one second is used:

select from trade where 
 int within hour 0D 0D00:01+2020.06.26D0 2020.06.26D07,
 otherTimeCol within 2020.06.26D0 2020.06.26D07

Better still would be to wrap this in a small utility for ease of use:

buffInts:{hour 0D 0D00:01+x}
select from trade where int within buffInts 2020.06.26D0 2020.06.26D07,
 otherTimeCol=2020.06.26D0 2020.06.26D07

Extended lookup table

Choosing a buffer value is an inexact science. A more efficient solution is to use a lookup table, this will allow for fast queries in both the hourly and fixed size partition examples. The table can be extended to include any extra columns as needed. .u.addLookup is edited to gather stats on the extra columns as needed:

.u.addLookup:{
 `:lookup/ upsert .Q.en[`:.] raze {select part:enlist x,tab:enlist y,
  minTS:min time,maxTS:max time,
  minOtherCol:min otherCol,maxOtherCol:max otherCol,
   from y}[x] each tables[]
 };

q)lookup
part tab minTS maxTS minOtherCol maxOtherCol
--------------------------------------------

cacheLookup behavior and the intLookup it creates are now also changed:

cacheLookup:{
 if[`lookup in tables[];
 intLookup::`lim xasc ungroup select column:`time`time`otherCol`otherCol,
             lim:(minTS,maxTS,minOtherCol,maxOtherCol) by part,tab from lookup;
 };

findInts:{[t;c;s;e]
 exec distinct part from intLookup where tab=t,column=c,lim within (s;e)
}

The user would then pass in an extra parameter to findInts to specify which column to use when choosing int partitions:

findInts[`quote;`otherCol;2020.06.28D16;2020.06.28D17]

These lookup tables are very powerful. Not only in these cases where data is slightly delayed but in fact any delay can now be handled gracefully, even if data is months late the lookup table protects against expensive full database scans or users missing data by making their queries too restrictive in their lookup of partitions assuming a certain maximum ‘lateness’ of data.

Reducing number of files

One side effect of int partitioning is a larger number of files being created on disk. At query time this can result in slower response time if many partitions need to be opened and scanned. Errors can also occur if the process ulimit is breached. At an extreme, the file system may run out of inode allocation space. Choosing how often a partition is created is one way to prevent too many files. Another is to implement a defrag process which will join several partitions together.

This process is started and passed the ports for the RDB and HDB:

q defrag.q -s 4 ::5011 ::5012

-s 4 is passed to create secondary threads so multiple cores are used to speed up the task.

The defrag function is then available which takes the following parameters:

• hdb – hsym to root of HDB
• src – int list of partitions to combine
• dst – int destination partition
• comp – compression settings
• p – symbol column name to apply parted attribute on
• typ – symbol hourly or fixed to specify the type of HDB

It’s source is viewable in defrag.q

Reducing hourly partitions

defrag[`:hourly/HDB;179608 179609;179608;17 2 6;`sym;`hourly]

For data to remain queryable in a performant manner there are some requirements:

• Partitions being joined must be contiguous
• The destination must be the minimum partition of the source list

These requirements are related to how the previously used hour function will be replaced. Now that the partitions are combined it will not function correctly:

q)select from trade where int in hour 2020.06.27D17, 
   time within 2020.06.27D17 2020.06.27D18
int sym time price size
-----------------------

This is due to the function expecting the data to be in partition 179609 which no longer exists as it has been merged in to 179608.

To solve this we can make use of the bin function. It returns the prevailing bucket a value falls in to:

q)list:0 2 4
q)list bin 0 1 3 5
0 0 1 2
q)list list bin 0 1 3 5
0 0 2 4

When kdb+ loads a partitioned database it creates a global variable which contains the list of all partitions. For date partitioned it is name date and for int int etc. This list can then be used to extend our hour function with bin to find the correct bucket:

q)hour:{int int bin `int$sum 24 1*`date`hh$\:x}

Our combined hours now correctly return that they both reside within a single partition:

q)hour 2020.06.27D16
179608
q)hour 2020.06.27D17
179608

The previously failing query now succeeds:

q)select from trade where int in hour 2020.06.27D17, 
   time within 2020.06.27D17 2020.06.27D18
int    sym time                          price    size
------------------------------------------------------
179608 baf 2020.06.27D17:00:00.000000000 0.949975 1   
179608 baf 2020.06.27D17:00:00.050000000 0.391543 2 

Reducing fixed partitions

defrag[`:fixed/HDB;0 1 2 3 4;0;17 2 6;`sym;`fixed]

The requirements about how partitions are combined which applied to hourly database do not apply to the fixed database. This is because the lookup table exists.

After running defrag no changes are needed in the helper function findInts. Instead during defrag the lookup table is updated with the latest information regarding partitions. To ensure two processes do not try to write to lookup simultaneously the RDB is contacted to perform this step.

The logic for this is best explored by viewing the reloadFixed function defined in defrag.q.

This blog has shown the flexibility int partitioning delivers by working through 2 examples. There are many other possible ways to exploit this feature. Some links to other examples are provided below.

• • https://code.kx.com/q/wp/intraday-writedown/ 
  • https://www.aquaq.co.uk/q/kdb-iot/ 
  • https://groups.google.com/g/personal-kdbplus/c/oe4UANpB1rM    

Full source code of this blog is available on Github.