Streaming functions introduction

Now for a moment let's take a break from the C++ API description (especially that it's a good spot, with all the types described), and talk about something new for version 1.1. I've been working on it in the background.

This new thing is the streaming functions. It's a cool and advanced concept, I've never seen it anywhere before, and for all I know I have invented it.

First let's look at the differences between the common functions and macros (or templates and such). Please turn your attention to the illustration below:

What happens during a function call? Some code (marked with the light bluish color) is happily zooming along when it decides to call a function. It prepares some arguments and jumps to the function code (reddish). The function executes, computes its result and jumps back to the point right after it has been called from. Then the original code continues from there (the slightly darker bluish color).

What happens during a macro (or template) invocation? It starts with some code zooming along in the same way, however when the macro call time comes, it prepares the arguments and then does nothing. It gets away with it because the compiler has done the work: it has placed the macro code right where it's called, so there is no need for jumps. After the macro is done, again it does nothing: the compiler has placed the next code to execute right after it, so it just continues on its way.

So far it's pretty equivalent. An interesting difference happens when the function or macro is called from more than one place. With a macro, another copy of the macro is created, inserted between its call and return points. That's why in the figure the macro is shown twice. But with the function the same function code is executed, and then returns back to the caller. That's why in the figure there are two function callers with their paths through the same function. But how does the function know, where should it jump on return? The caller tells it by pushing the return address onto the stack. When the function is done, it pops this address from the stack and jumps there.

Still, it looks all the same. A macro call is a bit more efficient, except when a large complex macro is called from many places, then it becomes more efficient as a function. However there is another difference if the function or macro holds some context (say, a static variable): each invocation of the macro will get its own context but all the function calls will share the same context. The only way to share the context with a macro is to pass some global context as its argument.

Now let's jump to the CEP world. The Sybase or StreamBase modules are essentially macros, and so are the Triceps templates. When such a macro gets instantiated, a whole new copy of it gets created with its tables/windows and streams/labels. Its input and output streams/labels get all connected in a fixed way. The limitation is that if the macro contains any tables, each instantiation gets a copy of it. Well, in Triceps you can use a table as an argument to a template In the other systems I think you still can't, so if you want to work with a common table in a module, you have to make up the query-response patterns, like the one described in the manual section "Comparative modularity".

In a query-response pattern there is some common sub-model, with a stream (in Triceps terms, a label, but here we're talking the other systems) for the queries to come in and a stream for the results to come out (both sides might have not only one but multiple streams). There are multiple inputs connected, from all the request sources, and the outputs are connected back to all the request sources. All the request sources (i.e. callers) get back the whole output of the pattern, so they need to identify, what output came from their input, and ignore the rest. They do this by adding the unique ids to their queries, and filter the results. In the end, it looks almost like a function but with much pain involved.

To make it look quite like a function, one thing is needed: the selective connection of the result streams (or, returning to the Triceps terminology, labels) to the caller. Connect the output labels, send some input, have it processed and send the result through the connection, disconnect the output labels. And what you get is a streaming function. It's very much like a common function but working on the streaming data arguments and results.

The next figure highlights the similarity and differences between the query patterns and the streaming functions.

The thick lines show where the data goes during one concrete call. The thin lines show the connections that do exist but without the data going through them at the moment (they will be used during the other calls, from these other callers). The dashed thin line shows the connection that doesn't exist at the moment. It will be created when needed (and at that time the thick arrow from the streaming to what is now the current return would disappear).

The particular beauty of the streaming functions for Triceps is that the other caller's don't even need to exist yet. They can be created and connected dynamically, do their job, call the function, use its result, and then be disposed of. The calling side in Triceps doesn't have to be streaming either: it could as well be procedural.

Types

Fundamentally, Triceps is a language, even though it is piggybacking on the other languages. And like in pretty much any programming language, pretty much anything in it has a type. Only the tip of that type system is exposed in the Perl API, as the RowType and TableType. But the C++ API has the whole depth. The types go all the way down to the simple types of the fields.

The classes for types are generally defined in the subdirectory type/. The class Type, defiined in type/Type.h is the common base class.

First, every kind of type has its entry in the enum TypeId:

        TT_VOID, // no value
        TT_UINT8, // unsigned 8-bit integer (byte)
        TT_INT32, // 32-bit integer
        TT_INT64,
        TT_FLOAT64, // 64-bit floating-point, what C calls "double"
        TT_STRING, // a string: a special kind of byte array
        TT_ROW, // a row of a table
        TT_RH, // row handle: item through which all indexes in the table own a row
        TT_TABLE, // data store of rows (AKA "window")
        TT_INDEX, // a table contains one or more indexes for its rows
        TT_AGGREGATOR, // user piece of code that does aggregation on the indexes
        TT_ROWSET, // an ordered set of rows

TT_ROWSET is something added in version 1.1.0, it will be described later. TT_VOID is pretty much a placeholder, in case if a void type would be needed later. The TypeId gets hardcoded in the constructor of every Type sub-class. It can be gotten back with the method

TypeId getTypeId() const;

Another method finds out if the type is the simple type of a field:

bool isSimple() const;

It would be true for the types of ids TT_VOID, TT_UINT8,  TT_INT32,  TT_INT64, TT_FLOAT64, TT_STRING.


Generally, you can check the TypeId and then cast the Type pointer to its subclass. All the simple types have the common base class SimpleType, which will be described in a moment.


There is also a static Type method that finds a simple type object by name (like "int32", "string" etc.):


static Onceref<const SimpleType> findSimpleType(const char *name);


Basically, there is not a whole lot of point in having lots of copies of the simple type objects (though if you want, you can). So there is one common copy of each simple type that can be found by name. If the type is known when you compile your C++ program, you can even avoid the look-up and refer to these objects directly.

    static Autoref<const SimpleType> r_void;
    static Autoref<const SimpleType> r_uint8;
    static Autoref<const SimpleType> r_int32;
    static Autoref<const SimpleType> r_int64;
    static Autoref<const SimpleType> r_float64;
    static Autoref<const SimpleType> r_string;

The type construction may cause errors. It is usually done either by a single constructor with all the needed arguments, or a simple constructor, then additional methods to add the information in bits in pieces, then an initialization method. In both cases there is a problem of how to report the errors. They're not easy to return from a constructor and a pain to check in the bit-by-bit construction.

Instead the error information gets internally collected in an Errors object, and can be read after the construction and/or initialization is completed:

virtual Erref getErrors() const;

A type with errors may not be used for anything other than reading the errors.

The rest of the common virtual methods has to do with the type comparison and print-outs. The comparison methods essentially check if two type objects are aliases for each other:

virtual bool equals(const Type *t) const;
virtual bool match(const Type *t) const;

The concept has been previously described with the Perl API. The equal types are exactly the same. The matching types are the same except for the names of their elements, so it's generally safe to pass the values between these types.

equals() is also available as operator==.

The print methods create a string representation of a type, used mostly for the error messages. There is no method to parse this string representation back, at least yet.

virtual void printTo(string &res, const string &indent = "", const string &subindent = "  ") const = 0;
string print(const string &indent = "", const string &subindent = "  ") const;

printTo() appends the information to an existing string. print() returns a new string with the message. print() is a wrapper around printTo() that creates an empty string, does printTo() into it and returns it.

The printing is normally done in a multi-line format, nicely indented, and the arguments indent and subindent define the initial indent level and the additional indentation for every level.

There is also a way to print everything in one line: pass the special constant NOINDENT (defined in common/StringUtil.h) in the argument indent. This is similar to using an undef for the same purpose in the Perl API.

The definitions of all the types are collected together in type/AllTypes.h.

Introduction to the C++ API

Fundamentally, the C++ and Perl APIs are shaped similarly. So I won't be making a version of all the examples in C++. Please read the Perl-based documentation first to understand the spirit and usage of Triceps. The C++-based documentation is more of the reference type and concentrates on the low-level specifics and differences from Perl necessitated by this specifics.

In many cases just reading the descriptions of the methods in the .h files should be enough to understand the details and be able to use the API. However in some cases the class hierarchies differ, with the Perl API covering the complexities exposed in the C++ API.

Writing directly in C++ is significantly harder than in Perl, so I highly recommend sticking with the Perl API unless you have a good reason to do otherwise. Even when the performance is important, it's usually enough to write a few critical elements in C++ and then bind them together in Perl. (If you wonder about Java, and why I didn't use it instead, the answer is that Java successfully combines the drawbacks of both and adds some of its own).

The C++ Triceps API is much more sensitive to the errors. The Perl API checks all the arguments for consistency, it's principle is that the interpreter must never crash. The C++ API is geared towards the efficiency of execution. It checks for errors when constructing the major elements but then does almost no checks at run time. The expectation is that the caller knows what he is doing. If the caller sends bad data, mislays the pointers etc., the program will crash. The idea here is that most likely the C++ API will be used from another layer: either an interpreted one (like Perl) or a compiled one (like a possible future custom language). Either way that layer is responsible for detecting the user errors at either interpretation or compile time. By the time the data gets to the C++ code, it's already checked and there is no need to check it again. Of course, if you write the programs manually in C++, that checking is upon you.

The more high-level features are currently available only in Perl. For example, there are no joins in the C++ API. If you want to do the joins in C++, you have to code your own. This will change over time, as these features will solidify and move to a C++ implementation to become more efficient and available through all the APIs. But it's much easier to experiment with the initial implementations in Perl.

The importance of DELETEs

Why did the last example worry so much about the DELETEs? Because without them, relying on just INSERTs for updates, it's easy to create bugs. The last example itself has an issue with handling the row replacement by INSERTs. Can you spot it from reading the code?

Here is run example that highlights the issue (as usual, the input lines are in italics):

OP_INSERT,1,AAA,10,10
Contents:
  id="1" symbol="AAA" price="10" size="10" 
tAvgPrice.out OP_INSERT symbol="AAA" id="1" price="10" 
OP_INSERT,3,AAA,20,20
Contents:
  id="1" symbol="AAA" price="10" size="10" 
  id="3" symbol="AAA" price="20" size="20" 
tAvgPrice.out OP_DELETE symbol="AAA" id="1" price="10" 
tAvgPrice.out OP_INSERT symbol="AAA" id="3" price="15" 
OP_INSERT,5,AAA,30,30
Contents:
  id="3" symbol="AAA" price="20" size="20" 
  id="5" symbol="AAA" price="30" size="30" 
tAvgPrice.out OP_DELETE symbol="AAA" id="3" price="15" 
tAvgPrice.out OP_INSERT symbol="AAA" id="5" price="25" 
OP_INSERT,5,BBB,30,30
Contents:
  id="5" symbol="BBB" price="30" size="30" 
tAvgPrice.out OP_INSERT symbol="BBB" id="5" price="30"
OP_INSERT,7,AAA,40,40
Contents:
  id="3" symbol="AAA" price="20" size="20" 
  id="7" symbol="AAA" price="40" size="40" 
tAvgPrice.out OP_DELETE symbol="AAA" id="5" price="25" 
tAvgPrice.out OP_INSERT symbol="AAA" id="7" price="30" 

The row with id=5 has been replaced to change the symbol from AAA to BBB. This act changes both the groups of AAA and of BBB, removing the row from the first one and inserting it into the second one. Yet only the output for BBB came out. The printout of the next row with id=7 and symbol=AAA shows that the row with id=5 has been indeed removed from the group AAA. It even corrects the result. But until that row came in, the average for the symbol AAA remained unchanged and incorrect.

There are multiple ways to fix this issue but first it had to be noticed. Which requires a lot of attention to detail. It's much better to avoid these bugs in the first place by sending the clean and nice input.

The index tree, part 3

Continuing our excursion into the internals of the table, the next example is of two parallel leaf index types:

TableType
+-IndexType A
+-IndexType B

The resulting internal arrangement is shown on Fig. 8.

Fig. 8 Two top-level index types.

Each index type produces exactly one index under the root (since the top-level index types always produce one index). Both indexes contain the same number of rows, and exactly the same rows. When a row is added to the table, it's added to all the leaf index types (one actual index of each type). When a row is deleted from the table, it's deleted from all the leaf index types. So the total is always the same. However the order of rows in the indexes may differ. The drawing shows the row references stacked in the same order as the index A because the index A is of the first leaf index type, and as such is the default one for the iteration.

The row handle contains the iterators for both paths, A and B. It's pretty normal to find a row through one index type and then iterate from there using the other index type.

The next example on Fig. 9 has a "primary" index with a unique key and a "secondary" index that groups the records:

TableType
+-IndexType A
+-IndexType B
  +-IndexType C

Fig. 9. A "primary" and "secondary" index type.

The index type A still produces one index and references all the rows directly. The index of type B produces the groups, with each group getting an index of type C. The total set of rows referrable through A and through B is still the same but through B they are split into multiple groups.

And Fig. 10 shows two leaf index types nested under one non-leaf.

TableType
+-IndexType A
  +-IndexType B
  +-IndexType C

Fig. 10. Two index types nested under one.

As usual, there is only one index of type A, and it splits the rows into groups. The new item in this picture is that each group has two indexes in it: one of type B and one of type C. Both indexes in the group contain the same rows. They don't decide, which rows they get. The index A decides, which rows go into which group. Then if the group 1 contains two rows, indexes B1 and C1, would both contain two rows each, the exact same set. The stack of row references has been visually split by groups to make this point more clear.

This happens to be a pretty useful arrangement: for example, B might be a hash index type, or in the future, a sorted index type, allowing to find the records by the key (and for the sorted index, to iterate in the order of keys), while C might be a FIFO index, keeping the insertion order, and maybe keeping the window size limited.

That's pretty much it for the basic index topologies. Some much more complex index trees can be created, but they would be the combinations of the examples shown. Also, don't forget that every extra index type adds overhead in both memory and CPU time, so avoid adding indexes that are not needed.

One more fine point has to do with the replacement policies. Consider that we have a table that contains the rows with the single field:

id int32

And the table type has two indexes:

TableType
+-IndexType "A" HashIndex key=(id)
+-IndexType "B" FifoIndex limit=3

Then we send there the rowops:

INSERT id=1
INSERT id=2
INSERT id=3
INSERT id=2

The second insert of id=2 triggers the replacement policy in both index types. In the index A it is a duplicate key and will cause the removal of the previous row with id=2. In the index B it overflows the limit and pushes out the oldest row, the one with id=1. If both records get deleted, the resulting table contents will be 2 rows (shown in FIFO order):

id=3
id=2

Which is probably not the best outcome. It might be tolerable with a FIFO index and a hashed index but gets even more annoying if there are two FIFO index types in the table: one top-level limiting the total number of rows, another one nested under a hashed index, limiting the number of rows per group, and they start conflicting this way with each other.

So the FIFO index is actually smart enough to avoid such problems: it looks at what the preceding indexes have decided to remove, checks if any of these rows belong to its group, and adjusts its calculation accordingly. In this example the index B will find out that the row with id=2 is already displaced by the index A. That leaves only 2 rows in the index B, so adding a new one will need no displacement. The resulting table contents will be

id=1
id=3
id=2

However here the order of index types is important. If the table were to be defined as

TableType
+-IndexType "B" FifoIndex limit=3
+-IndexType "A" HashIndex key=(id)

then the replacement policy of the index type B would run first, find that nothing has been displaced yet, and displace the row id=1. After that the replacement policy of the index type A will run, and being a hashed index, it doesn't have a choice, it has to replace the row id=2. And both rows end up displaced.

If the situations with automatic replacement of rows by the keyed indexes may arise, always make sure to put the keyed leaf index types before the FIFO leaf index types. However if you always diligently send a DELETE before the INSERT of the new version of the recond, then this problem won't occur and the order of index types will not matter.

The index tree, part 2

Let's use the Fig. 3 from the last post to go through how the other index-related operations work.

The iteration through the whole table starts with begin() or beginIdx(), the first being a form of the second that always uses the first leaf index type. BeginIdx() is fairly straightforward: it just follows the path from the root to the leaf, picking the first position in each index along the way, until it hits the RowHandle, as is shown in Fig. 4. That found RowHandle becomes its result. If the table is empty, it returns the NULL row handle.

If you specify a non-leaf index type as an argument of beginIdx(), the look-up can not just stop there because the non-leaf indexes are not directly connected to the row handles. It has to go through the whole chain to a leaf index. So, for a non-leaf index type argument the look-up is silently extended to its first leaf index type.

Fig. 4. Begin(), beginIdx($itA) and beginIdx($itB) work the same for this table.

The next pair is find() and findIdx(). As usual, find() is the same thing as findIdx() on the table's first leaf index type. It also follows the path from the root to the target index type. On each step it tries to find a matching position in the current index. If the position could not be found, the search fails and a NULL row handle is returned. If found, it is used to progress to the next index.

As has been mentioned before, the search always works internally on a RowHandle argument. If a plain Row is used as an argument, a new temporary RowHandle will be created for it, searched, and then freed after the search. This works well for two reasons. First, the indexes already have the functions for comparing two row handles to build their ordering. The same functions are reused for the search. Second, the row handles contain not only the index iterators but also the cached information from the rows, to make the comparisons faster. The exact kind of cached information varies by the index type. The FIFO indexes use none. The hashed indexes calculate a hash of the key field values, that will be used as a quick differentiator for the search. This information gets created when the row handle gets created. Whether the row handle is then used to insert into the table or to search in it, it's then used in the same way, to speed up the comparisons.

For findIdx(), the non-leaf index type arguments behave differently than the leaf ones: up to and including the index of the target type, the search works as usual. But then at the next level the logic switches to the same as in beginIdx(), going for the first row handle of the first leaf sub-index. This lets you find the first row handle of the matching group under the target index type.

If you use $table->findIdx($itA, $rh), on Fig. 5 it will go through the root index to the index A. There it will try to find the matching position. If none is found, the search ends and returns a NULL row handle. If the position is found, the search progresses towards the first leaf sub-index type. Which is the index type B, and which conveniently sits in this case right under A. The position in the index A determines, which index of type B will be used for the next step. Suppose it's the second position, so the second index of type B is used. Since we're now past the target index A, the logic used is the same as for beginIdx(), and the first position in B2 is picked. Which then leads to the first row handle of the second sub-stack of handles.

Fig. 5 FindIdx($itA, $rh) goes through A and then switches to the begin() logic.

The method firstOfGroupIdx() allows to navigate within a group, to jump from some row somewhere in the group to the first one, and then from there iterate through the group. The example of manual aggregation made use of it.

The Fig. 6 shows the example of $table->firstOfGroupIdx($itB, $rh), where $rh is pointing to the third record in B2. What it needs to do is go back to B2, and then execute the begin() logic from there on. However, remember, the row handle does not have a pointer to the indexes in the path, it only has the iterators. So, to find B2, the method does not really back up from the original row. It has to start back from the root and follow the path to B2 using the iterators in $rh. Since it uses the ready iterators, this works fast and requires no row comparisons. Once B2 (an index of type B) is reached, it goes for the first row in there.

FirstOfGroupIdx() works on both leaf and non-leaf index type arguments in the same way: it backs up from the reference row to the index of that type and executes the begin() logic from there. Obviously, if you use it on a non-leaf index type, the begin()-like part will follow its first leaf index type.

Fig. 6. FirstOfGroupIdx($itB, $rh)

The method nextGroupIdx() jumps to the first row of the next group, according to the argument index type. To do that, it has to retrace one level higher than firstOfGroupIdx(). Fig. 7 shows that $table->nextGroupIdx($itB, $rh) that starts from the same row handle as Fig. 6, has to logically back up to the index A, go to the next iterator there, and then follow to the first row of B3.

As usual, in reality there is no backing up, just the path is retraced from the root using the iterators in the row handle. Once the parent of index type B is reached (which is the index of type A), the path follows not the iterator from the row handle but the next one (yes, copied from the row handle, increased, followed). This gives the index of type B that contains the next group. And from there the same begin()-like logic finds its first row.

Same as firstOfGroupIdx(), nextGroupIdx() may be used on both the leaf and non-leaf indexes, with the same logic.

Fig. 7. NextGroupIdx($itB, $rh)

It's kind of annoying that firstOfGroupIdx() and nextGroupIdx() take the index type inside the group while findIdx() uses takes the parent index type to act on the same group. But as you can see, each of them follows its own internal logic, and I'm not sure if they can be reconciled to be more consistent.

At the moment the only navigation is forward. There is no matching last(), prev() or lastGroupIdx() or prevGroupIdx(). They are in the plan, but so far they are the victims of corner-cutting.

The index tree, part 1

The index types in a table type can form a pretty much arbitrary tree. Following the common tree terminology, the index types that have no other index types nested in them, are called the leaf index types. Since there seems to be no good one-word naming for the index types that have more index types nested in them ("inner"? "nested" is too  confusing), I simply call them non-leaf.

At the moment the Hashed index types can be used only in both leaf and non-leaf positions. The FIFO index types must always be in the leaf position, they don't allow the further nesting.

Now is the moment to look deeper into what is going on inside a table. Note that I've been very carefully talking about "index types" and not "indexes". In this post the difference matters. The index types are in the table type, the indexes are in the table. One index type may generate multiple indexes.

This will become clearer after you see the illustrations. I really hate doing the drawings, but here they are really necessary for the understanding, and way too complex for the ASCII-art. But first, the legend on Fig. 1.

Fig. 1. Drawings legend.

The nodes belonging to the table type are shown in red, the nodes belonging in the table are shown in blue, and the contents of the RowHandle is shown separately in yellow. The lines on the drawings represent not exactly pointers as such but more of the logical connections that may be more complicated than the simple pointers.

The lines in the RowHandle don't mean anything at all, they just show that the parts go together. In reality a RowHandle is a chunk of memory, with various elements placed in that memory. As far as indexes are concerned, the RowHandle contains an iterator for every index where it belongs. This lets it know its position in the table, to iterate along every index, and, most importantly, to be removed quickly from every index. A RowHandle belongs to one index of each index type, and contains the matching number of iterators in it.

The table type is shown as a normal flat tree. But the table itself is more complex and becomes 3-dimensional. Its "view from above" matches the table type's tree but the data grows "up" in the third dimension.

Let's start with the simplest case: a table type with only one index type. Whether the index type is hash or FIFO, doesn't matter here.

TableType
+-IndexType "A"

Fig. 2 shows the table structure.

Fig. 2. One index type.

The table here always contains exactly one index, matching the one defined index type, and the root index. The root index is very dumb, its only purpose is to tie together the multiple top-level indexes into a tree.

The only index of type A provides an ordering of the records, and this ordering is used for the iteration on the table.

For the next example let's look at the straight nesting at fig. 3.

TableType
+-IndexType "A"
  +-IndexType "B"

Fig. 3. Straight nesting.

The stack of row references is shown visually divided to match the indexing, but in reality there is no special division. This was done purely to make the picture easier to read.

There is still only one index of type A. And this is always the case with the top-level indexes, there is only one of them. This index divides the rows into 3 groups. Just like the rows in a leaf index, the groups in a non-leaf index are ordered in some way.

Each group then has its own second-level index of type B. Which then defines an order for the rows in it. To reiterate: the index of type A splits the rows by groups, then the group's index of type B defines the order of the rows in the group.

So what happens when we iterate through the table and ask for the next row handle? The current row handle contains the iterators in the indexes of types A and B. The easy thing is to advance the iterator of type B. Yeah, but in which index? The figure shows 3 indexes of type B, let's call them B1, B2 and B3. The iterator of type B in the row handle tells the relative position in the index, but it doesn't tell, which index it is. We need to step back and look at the index type A. It's the top-level index type, so there is always only one index for it. Then we take the iterator of type A and find this row's group in the index A. The group contains the index of type B, say B1. We can then take this index B1, take the iterator of type B from the row handle, and advance this iterator in this index. If the advance succeeded, then great, we've got the next row handle. But if the current row was the last row in B1, we need to step back to the index A again, advance the current row handle's iterator of type A there, find its index B2, and pick the first row handle of B2.

This process is what happens when we say $table->nextIdx($itB, $rh) or equivalently $rh->nextIdx($itB). The iteration goes by the leaf index type B, however it relies on all the index types in the path from the table type to B. If we do $rh->next(), the result is the same because the first leaf index type is used as the default index type for the iteration.

If we do $rh->next($itA),  the semantics is still the same: return the next row handle (not the next group). There is no way to get to the row handle without going all the way through a leaf index. So when a non-leaf index type is used for the iteration, it gets implicitly extended to its first nested leaf index type.

What would happen if a new row gets inserted, and the index type A determines that it does not belong to any of the existing groups? A new group will be created and inserted in the appropriate position in A's order. This group will have a new index of type B created, and the new row inserted in that index.

What would happen if both rows in B1 are removed? B1 will become empty and will be collapsed. The index A will delete the B1's group and B1 itself, and will remain with only two groups. The effect propagates upwards: if all the rows are removed, the last index of type B will collapse, then the index A will become empty and also collapse and be deleted. The only thing left will be the root index that stays in the existence no matter what.

When a table is first created, it has only the root index. The rest of the indexes pop into the existence as the rows get inserted. If you wonder, yes, this does apply to a table type with only one index type as well. Just this point has not been brought up until now.

Among all this froth of creation and collapse the iterators stay stable. Once a row is inserted, the indexes leading to it are not going anywhere (at least until that row gets removed). But since other rows and groups may be inserted around it, the notion of what row is next, will change over time.

CEP: why bother?

So far I've been writing largely in assumption that the readers would have an idea of what the Complex Event Processing is, and comparing to the other existing systems. It's a decent assumption for a quick start: if people have found Triceps, they've likely come through some CEP-related links, and know, what they are looking for.

But hopefully there will be novices as well, so let me have a go at describing the basics too.

If you look at Wikipedia, it has separate articles for the Event Stream Processing and the Complex Event Processing. In reality it's all the same thing, with the naming driven by the marketing. I would not be surprised if someone invents yet another name, and everyone will start jumping on that bandwagon too.

In general a CEP system can be thought of as a black box, where the input events come in, propagate in some way through that black box, and come out as the processed output events. There is also an idea that the processing should happen fast, though the definitions of "fast" vary widely.


If we open the lid on the box, there are at least three ways to think of its contents:

• a spreadsheet on steroids
• a data flow machine 
• a database driven by triggers

Hopefully you've seen a spreadsheet before. The cells in it are tied together by formulas. You change one cell, and the machine goes and recalculates everything that depends on it. So does a CEP system. If we look closer, we can discern the CEP engine (which is like the spreadsheet software), the CEP model (like the formulas in the spreadheet) and the state (like the current values in the spreadsheet).  An incoming event is like a change in an input cell, and the outgoing events are the updates of the values in the spreadsheet. Only a CEP system hand handle some very complicated formulas and many millions of records. There actually are products that connect the Excel spreadsheets with the behind-the-curtain computations in a CEP system, with the results coming back to the spreadsheet cells.  Pretty much every commercial CEP provider has a product that does that through the Excel RT interface. The way these models are written are not exactly pretty, but the results are, combining the nice presentation of spreadsheets and the speed and power of CEP.

A data flow machine, where the processing elements are exchanging messages, is your typical academical look at CEP. The events represented as data rows are the messages, and the CEP model describes the connections between the processing elements and their internal logic. This approach naturally maps to the multiprocessing, with each processing element becoming a separate thread. The hiccup is that the research in the dataflow machines tends to prefer the non-looped topologies. The loops in the connections complicate the things.

And many real-world relational databases already work very similarly to the CEP systems. They have constraints and triggers propagating the constraints. A trigger propagates an update on one table to an update on another table. It's like a formula in a spreasheet or a logical connection in a dataflow graph. Yet the databases usually miss two things: the propagation of the output events and the notion of "fast". The lack of propagation of the output events is totally baffling to me: the RDBMS engines already write the output event stream as the redo log. Why not send them also in some generalized format, XML or something? Then people realize that yes, they do want to get the output events and start writing some strange add-ons and aftermarket solutions like the log scrubbers. This has been a mystery to me for some 15 years. I mean, how more obvious can it be? But nobody budges. Well, with the CEP systems gaining popularity and the need to connect them to the databases, I think it will eventually grow on the database vendors that a decent event feed is a compatitive advantage, and I think it will happen somewhere soon. The felling of "fast" or lack thereof has to do with the databases being stored on disks. The growth of CEP has concided with the growth in RAM sizes, and the data is usually kept completely in memory. People who deploy CEP tend to want the performance not of hundreds or thousands but hundreds of thousands events per second. The second part of "fast" is connected with the transactions. In a traditional RDBMS a single event with all its downstream effects is one transaction. Which is safe but may cause lots of conflicts. The CEP systems usually allow to break up the logic into multiple loosely-dependent layers, thus cutting on the overhead.

How are the CEP systems used?

Despite what Wikipedia says, the pattern detection is NOT your typical usage, by a wide, wide margin. The typical usage is for the data aggregation: lots and lots of individual events come in, and you want to aggregate them to keep a concise and consistent picture for the decision-making.  The actual decision making can be done by humans or again by the CEP systems. The usages in the uses I know of vary from ad-click aggregation to the decisions to make a market trade, or watching whether the bank's end-of-day balance falls within the regulations.

A related use would be for the general alert consoles. The data aggregation is what they do too. The last time I looked up close (around 2006), the processing in the BMC Patrol and Nagios was just plain inadequate for anything useful, and I had to hand-code the data collection and console logic. But the CEP would have been just the ticket. I think, the only reaosn why it has not been widespread yet is that the commercial CEP licenses had cost a lot. But with the all-you-can-eat pricing of Sybase, and with the OpenSource systems, this is gradually changing.

Well, and there is also the pattern matching. It has been lagging behind the aggregation but growing too.

Issues with Triceps scheduling

Some of the issues have been already mentioned in the description of the loop scheduling: it's a bit confusing that the frame mark is placed on the next outer scheduling stack frame and not on the current one. This leads to the interesting effects in execution order.

But there are issues with the basic execution too:

The schedule() call, when used from inside the scheduled code, introduces unpredictability in the execution order: it puts the rowop after the last rowop in the outermost stack frame. But the outermost stack frame contains the queue of rowops that come from the outside. This means that the exact order of execution will depend on the timing of the rowops arriving from outside. Because of this, if the repeatable execution order is important, schedule() should be used only for the rowops coming from the outside.

The same issue happens with the loops that have been temporarily stopped and then resumed on arrival of more data from outside. The mark of such a loop will be unset when the loop continues, and looping at this mark will be equivalent to schedule(), having the same repeatability problem.

The call fork() is not exactly useful. Its main purpose was when I've thought that it's the solution to the problem of the loops. Which it has turned out to not solve, and another solution had to be devised. Now it really doesn't have much use, and will probably be removed in the future.

I have a few ideas for solutions of these issues, but they will need a bit more experimentation. Just keep in mind that the scheduling will be reformed in the future. It will still have the same general shape but differ in detail.

loop scheduling

The easiest way to schedule the loops is to do it procedurally, something like this:

foreach my $row (@rowset) {
  $unit->call($lbA->makeRowop(&Triceps::OP_INSERT, $row)); 
}

However the labels topologically connected into a loop can come handy as well. Some logic may be easier to express this way. Suppose the model contains the labels connected in a loop:

X->A->B->C->Y

   ^     |

   +-----+

Suppose a rowop X1 is scheduled for label X, and causes the loop executed twice, with rowops X1, A2, B3, C4, A5, B6, C7, Y8. If each operation is done as a CALL, the stack grows like this: It starts with X1 scheduled.

[X1]

Which then gets executed, with its own execution frame (marked as such for clarity:

[ ] of X1
[ ]

Which then calls A2:

[ ] of A2
[ ] of X1
[ ]

By the time the execution comes to Y8, the stack looks like:

[ ] of Y8
[ ] of C7
[ ] of B6
[ ] of A5
[ ] of C4
[ ] of B3
[ ] of A2
[ ] of X1
[ ]

The loop has been converted into recursion, and the whole length of execution is the deep of the recursion. If the loop executes a million times, the stack will be two million levels deep. Worse yet, it's not just the Triceps scheduler stack that grows, it's also the process (C++) stack.

Would things be better with FORK instead of CALL used throughout the loop? It starts the  same way:

[X1]

Then X1 executes, gets its own frame and forks A2:

[A2] of X1
[ ]

Then A2 executes, gets its own frame and forks B3:

[B3] of A2
[ ] of X1
[ ]

By the end of the loop the picture becomes exactly the same as with CALL. For a while I've thought that optimizing out the empty stack frames would solve the problem, but no, that doesn't work: the problem is that the C++ process stack keeps growing no matter what. The jump back in the loop needs to be placed into an earlier stack frame.

One way to do it would be to use the SCHEDULE operation in C to jump back to A, placing the rowop A5 back onto the outermost frame. The scheduler stack at the end of C4 would look like:

[ ] of C4
[ ] of B3
[ ] of A2
[ ] of X1
[A5]

Then the stack would unwind back to

[A5]

and the next iteration of the loop will start afresh. The problem here is that if X1 wanted to complete the loop and then do something, it can't. By the time the second iteration of the loop starts, X1 is completely gone. It would be better to be able to enqueue the next execution of the loop at the specific point of the stack.

Here the concept of the frame mark comes in: a frame mark is a token object, completely opaque to the program. It can be used only in two operations:

• setMark() remembers the position in the frame stack, just outside the current frame
• loopAt() enqueues a rowop at the marked frame

Then the loop wold have its mark object M. The label A will execute setMark(M), and the label C will execute loopAt(M, rowop(A)). The rest of the execution can as well use call().

When A2 calls setMark(M), the stack will look like this:

[ ] of A2
[ ] of X1 * mark M
[ ]

The mark M remembers the frame one outer to the current one. The stack at the end of C4, after it has called loopAt(M, A5), is:

[ ] of C4
[ ] of B3
[ ] of A2
[A5] of X1 * mark M
[ ]

The stack then unwinds until A5 starts its execution:

[ ] of A5
[ ] of X1 * mark M
[ ]

Each iteration starts with a fresh stack, and the stack depth is limited to one iteration. The nested loops can also be properly executed.

Now, why does the mark is placed on the frame that is one out from the current one? Suppose that it did remember the current frame. Then at the end of C4 the stack will be:

[ ] of C4
[ ] of B3
[A5] of A2 * mark M
[ ] of X1
[ ]

The stack will unwind until A5. Which would then have its own frame pushed onto the stack, and call setMark(M) again, moving the mark to its own frame:

[ ] of A5 * mark M
[ ] of A2
[ ] of X1
[ ]

So on each iteration of the loop one extra frame will be pushed onto the stack, and the mark moved by one level. A loop executing a million times will push a million frames, which is bad. Marking the next outer frame prevents this.  Another option would have been to put the mark in X, but that would mean that every loop must have a preceding label that just marks the frame (well, and potentially could do the other initializations too), which seems to be too annoying.

However as things are, another problem is that if X does call(A2), when it returns, the loop would not be completed yet, only the first iteration would be completed. To have the whole loop completed, there would have to be another label W, and when W does call(X1), the loop would be completed.

This is still messy, and I'm still thinking about the ways to improve the situation.

What happens after the stack unwinds past the mark? The mark gets unset. When someone calls loopAt() with an unset mark, the rowop is enqueued in the outermost frame, having the same effect as schedule().

rowop sets the condition, it would free that original row and make it continue through the loop.  Eventually the loop will come to the looping point, calling loopAt(). But the original mark will be long unset. Scheduling at the outermost frame seems to be a logical thing to do at this point.

What if setMark() is called when there is only one frame on the stack? Then there is no second frame outer to it. The mark will simply be left unset.

Basic scheduling

The Triceps scheduler provides 3 basic ways of executing of a label:

• Call: execute the label right now, including all the nested calls. All of this will be completed after the call returns.
• Fork: execute the label after the current label returns but before anything else is done. Obviously, if multiple labels are forked, they will execute in order after the current label returns (but before its caller gets the control back).
• Schedule: execute the label after everything else is done.

How it works inside:

A scheduler in the execution unit keeps a stack of queues. Each queue is essentially a stack frame. The stack always contains at least one queue, which is called the outermost stack frame. 

When the new rowops come from the outside world, they are added with schedule() to that stack frame. That's what schedule() does: always adds messages to the outermost stack frame. If rowops 1, 2 and 3 are added, the stack looks like this:

[1, 2, 3]

The unit method drainFrame() is then used to process the rowops. It makes the unit call each label on the innermost frame (which is at the moment the same as outermost frame) in order.

First it calls the rowop 1. It's removed from the queue, then a new frame is pushed onto the stack

[ ]
[2, 3]

Then the rowop 1 executes. If it calls rowop 4, another frame is pushed onto the stack

[ ]
[ ]
[2, 3]

then the rowop 4 executes. After it is finished (not calling any other rowops), the outermost empty frame is popped before the execution of rowop 1 continues.

[ ]
[2, 3]


Suppose then rowop 1 forks rowops 5 and 6. They are appended to the innermost frame.

[5, 6]
[2, 3]

If rowop 1 calls rowop 7, again a frame is pushed onto the stack before it executes

[ ]
[5, 6]
[2, 3]


Again, note that a call does not put the target rowop into any scheduler queue. The identity of rowop being processed is just kept in the call context. A call also involves a direct C++ call on the thread stack, and if any Perl code is involved, a Perl call too. Because of this, if you nest the calls too deeply, you may run out of the thread stack space.

Returning back to the sequence, after the call of rowop 7 completes, the scheduler stack returns to

[5, 6]
[2, 3]


Suppose now the execution of rowop 1 completes. But its call is not done yet, because its stack queue is not empty. The scheduler calls drainFrame() recursively, which picks the next rowop from the innermost queue (rowop 5), and calls it, pushing a new stack frame and executing the rowop 5 code:

[ ]
[6]
[2, 3]

If rowop 5 forks rowop 8, the stack becomes:

[8]
[6]
[2, 3]

When the execution of rowop 5 returns, its queue is also not empty. So the scheduler calls rowop 8. During its execution the stack is:

[ ]
[ ]
[6]
[2, 3]

Suppose the rowop 8 doesn't call or fork anything else and returns. Its innermost queue is empty, so the call completes and pops the stack frame.

[ ]
[6]
[2, 3]

Now the queue of rowop 5 is also empty, so its call completes and pops the stack frame.

[6]
[2, 3]

The call of rowop 6 begins.

[ ]
[ ]
[2, 3]

Suppose rowop 6 calls schedule() of rowop 9. Rowop 9 is then added to the outermost queue:

[ ]
[ ]
[2, 3, 9]

Rowop 6 then returns, its queue is empty, so it's popped and its call completes.

[ ]
[2, 3, 9]

Now the queue of rowop 1 has become empty, so it's popped from the stack and the call fo rowop 1 completes.

[2, 3, 9]

The unit method drainFrame() keeps running, now taking the rowop 2 and executing it, and so on, until the outermost queue becomes empty, and drainFrame() returns.

No bundling

The most important principle of Triceps scheduling is: No Bundling. Every rowop is for itself. The bundling is what messes up the Coral8 scheduler the most.


What is a bundle? It's a set of records that go through the execution together. If you have two functional elements F1 and F2 arranged in a sequential fashion F1-> F2, and a few loose records R1, R2, R3, the normal execution order will be:


F1(R1), F2(R1),
F1(R2), F2(R2),
F1(R3), F2(R3)

If the same records are placed in a bundle, the execution order will be different:


F1(R1), F1(R2), F1(R3),
F2(R1), F2(R2), F2(R3)

That would not be a problem, and even could be occasionally useful, if the bundles were always created explicitly. In reality every time a statement produces multiple record from a single one (think of a join that picks multiple records from another side), it creates a bundle and messes up all the logic after it. Some logic gets affected so badly that a few statements in CCL (like ON UPDATE) had to be designated as always ignoring the bundles, otherwise they would not work at all. At DB I wrote a CCL pattern for breaking up the bundles. It's rather heavyweight and thus could not be used all over the place but provides a generic solution for the most unpleasant cases.

Worse yet, the bundles may get created in Coral8 absolutely accidentally: if two records happen to have the same timestamp, for all practical purposes they would act as a bundle. In models that were designed without the appropriate guards, this leads to the time-based bugs that are hard to catch and debug. Writing these guards correctly is hard, and testing them is even harder.

Another issue with bundles is that they make the large queries slower. Suppose you do a query from a window that returns a million records.  All of them will be collected in a bundle, then the bundle will be sent to the interface gateway that would build one huge protocol packet, which will then be sent to the client, which will receive the whole packet and then finally iterate on the records in it. Assuming that nothing runs out of memory along the way, it will be a long time until the client sees the first record.  Very, very annoying.

Aleri also has its own version of bundles, called transactions, but a more smart one. Aleri always relies on the primary keys. The condition for a transaction is that it must never contain multiple modification for the same primary key. Since there are no execution order guarantees between the functional elements, in this respect the transactions work in the same way as loose records, only with a more efficient communication between threads. Still, if the primary key changes in an element, the condition does not propagate through it. Such elements have to internally collapse the outgoing transactions along the new key, adding overhead.

Introduction to Triceps scheduling

The main point of an execution unit in Triceps is scheduling of the execution of the row operations. It keeps a queue of the operations and selects, which one to execute next. The scheduling is important for a predictable execution order within a single thread.

There are multiple approaches to scheduling. Aleri essentially doesn't have any, except for the flow control between threads, because each its element is a separate thread. Coral8 has an intricate scheduling algorithm. Sybase R5 has the same logic as Coral8 inside each thread. StreamBase presumably also has some.

The scheduling logic in Triceps is different from the other CEP systems. The Coral8 logic looks at first like the only reasonable way to go, but could not be used for three reasons: First, it's a trade secret, so it can't be simply reused. If I'd never seen it, that would not be an issue but I've worked on it and implemented its version for R5. Second, it relies on the properties that the compiler computes from the model graph analysis. Triceps has no compiler, and could not do this. Third, in reality it simply doesn't work that well. There are quite a few cases when the Coral8 scheduler comes up with a strange and troublesome execution order.

For a while I've hoped that Triceps would need no scheduler at all, and everything would be handled by the procedural calls.This has proved to have its own limitations, and thus the labels and their scheduling were born. The Triceps scheduling still has issues to resolve, but overall still feels much better than the Coral8 one.

a little about templates

Since people have started commenting about templates, let me show a bit more, what do I mean by them on a simple example.

Coral8 doesn't provide a way to query the windows directly, especially when the CCL is compiled without debugging. So you're expected to make your own. People at DB have developed a nice pattern that goes approximately like this:

// some window that we want to make queryable
create window w_my schema s_my
keep last per key_a per key_b
keep 1 week;

// the stream to send the query requests
// (the schema can be shared by all simple queries) 
create schema s_query (
  qqq_id string // unique id of the query
);
create input stream query_my schema s_query;

// the stream to return the results
// (all result streams will inherit a partial schema)
create schema s_result (
  qqq_id string, // returns back the id received in the query
  qqq_end boolean, // will be TRUE in the special end indicator record
);
create output stream result_my schema inherits from s_result, s_my;

// now process the query
insert into result_my
select q.qqq_id, NULL, w.*
from s_query as q, w_my as w;

// the end marker
insert into result_my (qqq_id, qqq_end)
select qqq_id, TRUE
from s_query;

To query the window, a program would select a unique query id, subscribe to result_my with a filter (qqq_id = unique_id) and send a record of (unique_id) into query_my.  Then it would sit and collect the result rows. Finally it would get a row with qqq_end = TRUE and disconnect.

This is  a fairly large amount of code to be repeated for every window. What I would like to to instead is to just write

create window w_my schema s_my
keep last per key_a per key_b
keep 1 week;
make_queryable(w_my);

and have the template make_queryable expand into the rest of the code (obviously, the schema definitions would not need to be expanded repeatedly, they would go into an include file).

To make things more interesting, it would be nice to have the query filter the results by some field values. Nothing as fancy as SQL, just by equality to some fields. Suppose, s_my includes the fields field_c and field_d, and we want to be able to filter by them. Then the query can be done as:

create input stream query_my schema inherits from s_query (
  field_c integer,
  field_d string
);

// result_my is the same as before...

// query with filtering (in a rather inefficient way) 
insert into result_my
select q.qqq_id, NULL, w.*
from s_query as q, w_my as w
where
  (q.field_c is null or q.field_c = w.field_c)
  and (q.field_d is null or q.field_d = w.field_d);

// the end marker is as before
insert into result_my (qqq_id, qqq_end)
select qqq_id, TRUE
from s_query;

It would be nice then to create this kind of query as a template instantiation

make_query(w_my, (field_c, field_d));

If there weren't already an entrenched tradition at DB, I would not write directly in CCL at all. I would have made a macro language that would generate CCL. Of course, then the IDE would see only the results of the code generation and could not be used directly to write code in it, but who cares, IDEs are useless for this purpose anyway.

Interestingly, there already are people who do that kind of things. Some people actually prefer the Aleri XML format because it's easier for them to generate the code in XML. (I don't exactly see why generating the code in XML would be easier but there are all kinds of weird XML-based infrastructures out there).

printing the object contents

When debugging the programs, it's important to find from the error messages, what is going on, what kinds of objects are getting involved. Because of this, most of the Triceps objects provide a way to print out their contents into a string. This is done with the method print(). The simplest use is as follows:

$message = "Error in object " . $object->print();

Most of the objects tend to have a pretty complicated internal structure and are printed on multiple lines. They look better when the components are appropriately indented. The default call prints as if the basic message is un-indented, and indents every extra level by 2 spaces.

This can be changed with extra arguments. The general format of print() is:

$object->print([indent, [subindent] ])

where indent is the initial indentation, and subindent is the additional indentation for every level. So the default print() is equivalent to print("", "  ").

A special case is

$object->print(undef)

It prints the object in a single line, without line breaks.

The row types support the print() method. Here is an example of how a type would get printed:

$rt1 = Triceps::RowType->new(
  a => "uint8",
  b => "int32",
  c => "int64",
  d => "float64",
  e => "string",
);

Then $rt1->print() produces:

row {
  uint8 a,
  int32 b,
  int64 c,
  float64 d,
  string e,
}

With extra arguments $rt1->print("++", "--"):

row {
++--uint8 a,
++--int32 b,
++--int64 c,
++--float64 d,
++--string e,
++}

And finally with an undef argument $rt1->print(undef):

row { uint8 a, int32 b, int64 c, float64 d, string e, }

API introduction

As mentioned before, at the moment Triceps provides the APIs in C++ and Perl. They are similar but not quite the same, because the nature of the compiled and scripted languages is different. The C++ API is more direct and expects discipline from the programmer: if some incorrect arguments are passed, everything might crash. The Perl API should never crash. It should detect any incorrect use and report an orderly error. Besides, the idioms of the scripted languages are different from the compiled languages, and different usages become convenient.


The Perl API is implemented in XS. Some people, may wonder, why not SWIG? SWIG would automatically export the API into many languages, not just Perl. The problem with SWIG is that it just maps the API one to one.  And this doesn't work any good, it makes some very ugly APIs with abilities to crash. Which then have to be wrapped into more scripting code before they become usable. So then why bother with SWIG, easier to just use the scripting language's native extension methods. Another benefit of the native support is the access to the correct memory management.

In general, I've tried to avoid the premature optimization. The idea is to get it working at all first, and then bother about working fast. Except for the cases when the need for optimization looked obvious, and the logic intertwined with the general design strongly ehough, that if done one way, would be difficult to change in the future. We'll see, if these "obvious" cases really turn out to be the obvious wins, or will they become a premature-optimization mess.

Going forward, I'll try to split the subjects into the separate posts, centered around Perl API, C++ API, or guts of the (C++) implementation and tag them as such. Feel free to pick only the ones you're interested in.Some of the posts still will be general, and have all three tags.

Hello, world!

Let's finally get to business: write the "Hello, world!" program with Triceps.  Since Triceps is an embeddable library, naturally, the smallest "Hello, world!" program would be in the host language without Triceps, but it would not be interesting. So here is the a bit contrived but more interesting Perl program that passes some data through the Triceps machinery:

use Triceps;

$hwunit = Triceps::Unit->new("hwunit") or die "$!";
$hw_rt = Triceps::RowType->new(
  greeting => "string",
  address => "string",
) or die "$!";

my $print_greeting = $hwunit->makeLabel($hw_rt, "print_greeting", undef, 
  sub {
    my ($label, $rowop) = @_;
    printf "%s!\n", join(', ', $rowop->getRow()->toArray());
  } 
) or die "$!";

$hwunit->call($print_greeting->makeRowop(&Triceps::OP_INSERT,
  $hw_rt->makeRowHash(
    greeting => "Hello",
    address => "world",
  ) 
)) or die "$!";

What happens there? First, we import the Triceps module. Then we create a Triceps execution unit. An execution unit keeps the Triceps context and controls the execution for one thread. Nothing really stops you from having multiple execution units in the same thread, however there is not a whole lot of benefit in it either. But a single execution unit must never ever be used in multiple threads. It's single-threaded by design and has no synchronization in it. The argument of the constructor is the name of the unit, that can be used in printing messages about it. It doesn't have to be the same as the name of the variable that keeps the reference to the unit, but it's a convenient convention to make the debugging easier.

If something goes wrong, the constructor will return an undef and set the error message in $!. This actually has turned out to be not such a good idea as it seemed, since writing "or die" at every line quickly turns tedious. And there is usually not much point in catching the errors of this type, since they are essentially the compilation errors and should cause the program to die anyway. So, this will be soon changed throughought the code to just die with the message (and if it needs to be caught, it can be caught with eval).

The next statement creates the type for rows. For the simplest example, one row type is enough. It contains two string fields. A row type does not belong to an execution unit. It may be used in parallel by multiple threads. Once a row type is created, it's immutable, and that's the story for pretty much all the Triceps objects that can be shared between multiple threads: they are created, they become immutable, and then they can be shared. (Of course, the containers that facilitate the passing of data between the threads would have to be an exception to this rule).

Then we create a label. If you look at the "SQLy vs procedural" example a little while back, you'll see that the labels are analogs of streams in Coral8. And that's what they are in Triceps. Of course, now, in the days of the structured programming, we don't create labels for GOTOs all over the place. But we still use labels. The function names are labels, the loops in Perl may have labels. So a Triceps label can often be seen kind of like a function definition, but so far only kind of. It takes a data row as a parameter and does something with it. But unlike a proper function it has no way to return the processed data back to the caller. It has to either pass the processed data to other labels or collect it in some hardcoded data structure, from which the caller can later extract it back. This means that until this gets worked out better, a Triceps label is still much more like a GOTO label or Coral8 stream than a proper function. Just like the unit, a label may be used in only one thread.

A label takes a row type for the rows it accepts, a name (again, purely for the ease of debugging) and a reference to a Perl function that will be processing the data. Extra arguments for the function can be specified as well, but there is no use for them in this example.

Here it's a simple unnamed function. Though of course a reference to a named function can be used instead, and the same function may be reused for multiple labels. Whenever the label gets a row operation to process, its function gets called with the reference to the label object, the row operation object, and whatever extra arguments were specified at the label creation (none in this example). The example just prints a message combined from the data in the row.

Note that a label doesn't just get a row. It gets a row operation ("rowop" as it's called throughout the code). It's an important distinction. A row just stores some data. As the row gets passed around, it gets referenced and unreferenced, but it just stays the same until the last reference to it disappears, and then it gets destroyed. It doesn't know what happens with the data, it just stores them. A row may be shared between multiple threads. On the other hand, a row operation says "take these data and do a such and such thing with them".  A row operation is a combination of a row of data, an operation code, and a label that is to execute the operation. It is confined to a single thread. Inside this thread a reference to a row operation may be kept and reused again and again, since the row operation object is also immutable.


Triceps has the explicit operation codes, very much like Aleri (only Aleri doesn't differentiate between a row and row operation, every row there has an opcode in it, and the Sybase CEP R5 does the same). It might be just my background, but let me tell you: the CEP systems without the explicit opcodes are a pain. The visible opcodes make life a lot easier. However unlike Aleri, there is no UPDATE opcode. The available opcodes are INSERT, DELETE and NOP (no-operation). If you want to update something, you send two operations: first DELETE for the old value, then INSERT for the new value. There will be a section later with more details and comparisons, but for now that's enough information.

For this simple example, the opcode doesn't really matter, so the label function quietly ignores it. It gets the row from the row operation and extracts the data from it into the Perl format, then prints them. There are two Perl formats supported: an array and a hash. In the array format, the array contains the values of the fields in the order they are defined in the row type. The hash format consists of name-value pairs, which may be stored either in an actual hash or in an array. The conversion from row to a hash actually returns an array of values which becomes a hash if it gets stored into a hash variable.

As a side note, this also suggests, how the systems without explicit opcodes came to be: they've been initially built on the  simple stateless examples. And when the more complex examples have turned up, they've been aready stuck on this path, and could not afford too deep a retrofit.

The final part of the example is the creation of a row operation for our label, with an INSERT opcode and a row created from hash-formatted Perl data, and calling it through the execution unit. The row type provides a method to construct the rows, and the label provides a method to construct the row operations for it. The call() method of the execution unit does exactly what its name implies: it evaluates the label function right now, and returns after all its processing its done.

Enter Triceps

The Triceps development has been largely shaped by two considerations:

It has to be different from the Sybase products on which I worked. This is helpful from both legal standpoint and from marketing standpoint: Sybase and StreamBase already have similar products that compete head to head. There is no use getting into the same fray without major resources.

It has to be small. I can't spend the same amount of effort on Triceps as a large company, or even as a small one. Not only this saves time but also allows the modifications to be easy and fast. The point of Triceps is to experiment with the CEP language to make it easy to use: try out the ideas, make sure that they work well, or replace them with other ideas. The companies with a large established product can't really afford the radical changes: they have invested much effort into the product, and are stuck with supporting it and providing compatibility into the future.

Both of these considerations point into the same direction: an embeddable CEP system. Adapting an integrated system for an embedded usage is not easy, so it's a good open niche. Yeah, there is Esper, but from a cursory look, it seems to have the same issues as Coral8/StreamBase.

And an embeddable system saves on a lot of components.

For starters, no IDE. Anyway, I find the IDEs pretty useless for development in general, and especially for the CEP development. Though it comes handy once in a while for the analysis of the code and debugging.

No new language, no need to develop compilers, virtual machines, function libraries, external callout APIs. Well, the major goal of Triceps is actually the development of a new and better language. But it's one of these paradoxes: Aleri does the relational logic looking like procedural, Coral8 and StreamBase do the procedural logic looking like relational, and Triceps is a design of a language without a language. Eventually there probably will be a language, to be mixed with the parent one. But for now a lot can be done by simply using the Triceps library in an existing scripting language. The existing scripting languages are already powerful, fast, and also allow the dynamic compilation.

No separate server executable, no need to control it, and no custom network protocols: the users can put the code directly into their executables and devise any protocols they please. Well, it's not a real good answer for the protocols, since it means that everyone who wants to communicate the streaming data for Triceps over the network has to implement these protocols from scratch. So eventually Triceps will provide a default implementation. But it doesn't have to be done right away.

No data persistence  for now either. It's a nice feature, and I have some ideas about it too, but it requires a large amount of work, and doesn't really affect the API.

The language used to implement Triceps is C++, and the scripting language is Perl. Nothing really prevents embedding Triceps into other languages but it's not going to happen anywhere soon. The reason being that extra code adds weight and makes the changes more difficult.

The multithreading support has been a major consideration from the start. All the C++ code has been written with the multithreading in mind. However for the first release the multithreading did not propagate into the Perl API yet.

Even though Triceps is an experimental system, that does not imply that it's of a toy quality. The code is written in production quality to start with, with a full array of unit tests.

we're not in 1950s any more, or are we?

Part of the complexity with CCL programming is that the CCL programs tend to feel very broken-up, with the flow of the logic jumping all over the place.

Consider a simple example: some incoming financial information may identify the securities by either RIC (Reuters identifier) or SEDOL or  ISIN, and before processing it further we want to convert them all to ISIN (since the fundamentally same security may be identified in multiple ways when it's traded in multiple countries).

This can be expressed in CCL approximately like this (no guarantees about the correctness of this code, since I don't have a compiler to try it out):

// the incoming data
create schema s_incoming (
  id_type string, // identifier type: RIC, SEDOL or ISIN
  id_value string, // the value of the identifier
  // add another 90 fields of payload...
);

// the normalized data
create schema s_normalized (
  isin string, // the identity is normalized to ISIN
  // add another 90 fields of payload...
);

// schema for the identifier translation tables
create schema s_translation (
  from string, // external id value (RIC or SEDOL)
  isin string, // the translation to ISIN
);

// the windows defining the translations from RIC and SEDOL to ISIN
create window w_trans_ric schema s_translation
  keep last per from;
create window w_trans_sedol schema s_translation
  keep last per from;

create input stream i_incoming schema s_incoming;
create stream incoming_ric  schema s_incoming;
create stream incoming_sedol  schema s_incoming;
create stream incoming_isin  schema s_incoming;
create output stream o_normalized schema s_normalized;

insert
  when id_type = 'RIC' then incoming_ric
  when id_type = 'SEDOL' then incoming_sedol
  when id_type = 'ISIN' then incoming_isin
select *
from i_incoming;

insert into o_normalized
select
  w.isin,
  i. ... // the other 90 fields
from
  incoming_ric as i join w_tranc_ric as w
    on i.id_value =  w.from;

insert into o_normalized
select
  w.isin,
  i. ... // the other 90 fields
from
  incoming_sedol as i join w_tranc_sedol as w
    on i.id_value =  w.from;

insert into o_normalized
select
  i.id_value,
  i. ... // the other 90 fields
from
  incoming_isin; 

Not exactly easy, is it, even with the copying of payload data skipped? You may notice that what it does could also be expressed as procedural pseudo-code:

// the incoming data
struct s_incoming (
  string id_type, // identifier type: RIC, SEDOL or ISIN
  string id_value, // the value of the identifier
  // add another 90 fields of payload...
);

// schema for the identifier translation tables
struct s_translation (
  string from, // external id value (RIC or SEDOL)
  string isin, // the translation to ISIN
);

// the windows defining the translations from RIC and SEDOL to ISIN
table s_translation w_trans_ric
  key from;
table s_translation w_trans_sedol
  key from;

s_incoming i_incoming;
string isin;

if (i_incoming.id_type == 'RIC') {
  isin = lookup(w_trans_ric, 
    w_trans_ric.from == i_incoming.id_value
  ).isin;
} elsif (i_incoming.id_type == 'SEDOL') {
  isin = lookup(w_trans_sedol, 
    w_trans_sedol.from == i_incoming.id_value
  ).isin;
} elsif (i_incoming.id_type == 'ISIN') {
  isin = i_incoming.id_value;
}

if (isin != NULL) {
  output o_ normalized(isin,
    i_incoming.(* except id_type, id_value)
  );
}

Basically, writing in CCL feels like programming in Fortran in the 50s: lots of labels, lots of GOTOs. Each stream is essentially a label, when looking from the procedural standpoint. It's actually worse than Fortran, since all the labels have to be pre-defined (with types!). And there isn't even the normal sequential flow, each statement must be followed by a GOTO, like on those machines with magnetic-drum main memory.

This is very much like the example in my book, in section 6.4. Queues as the sole synchronization mechanism. You can alook at the draft text online. This similarity is not accidental: the CCL streams are queues, and they are the only communication mechanism in CCL.

The SQL statement structure also adds to the confusion: each statement has the destination followed by the source of the data, so each statement reads like it flows backwards.

In Triceps I aim to do better. It is not as smooth as the shown pseudo-code yet, but things are moving in this direction. I have a few ideas about improving this pseudo-code too but they would have to wait until another day.

P.S. I don't seem to be able to post comments. I'm not sure, what is wrong with the Blogspot engine. But answering the comment, yeah, I don't know much about Esper. Both Coral8 and Streambase also have the .* syntax, and Aleri has a similar ExtendStream. However that copies all the fields, without dropping any of them (like id_type and id_value here).

surveying the landscape

What do we have in the CEP area now? The scene is pretty much dominated by Sybase (combining the former Aleri and Coral8) and StreamBase.

There seem to be two major approaches to the execution model. One was used by Aleri, another by Coral8 and StreamBase. I'm not hugely familiar with StreamBase, but that's how it seems to me. Since I'm much more familiar with Coral8, I'll be calling the second model the Coral8 model. If you find StreamBase substantially different, let me know.

The Aleri idea is to collect and keep all the data. The relational operators get applied on the data, producing the derived data ("materialized views") and eventually the results. So, even though the Aleri models were usually expressed in XML (though an SQL compiler was also available), fundamentally it's a very relational and SQLy approach.

This creates a few nice properties. All steps of execution can be pipelined and executed in parallel.For persistence, it's fundamentally enough to keep only the input data (what has been called BaseStreams and then SourceStreams), and all the derived computations can be easily reprocessed on restart (it's funny but it turns out that often it's faster to read a small state from the disk and recalculate the rest from scratch in memory than to load a large state from the disk).

It also has issues. It doesn't allow loops, and the procedural calculation aren't always easy to express. And keeping all the state requires more memory. The issues of loops and procedural computations have been addressed by FlexStreams: modules that would perform the procedural computations instead of relational operations, written in SPLASH - a vaguely C-ish or Java-ish language. However this tends to break the relational properties: once you add a FlexStream, usually  you do it for the reasons that prevent the derived calculations from being re-done, creating issues with saving and restoring the state. Mind you, you can write a FlexStream that doesn't break any of them, but then it would probably be doing something that can be expressed without it in the first place.

Coral8 has grown from the opposite direction: the idea has been to process the incoming data while keeping a minimal state in variables and short-term "windows" (limited sliding recordings of the incoming data). The language (CCL) is very SQL-like. It relies on the state of variables and windows being pretty much global (module-wide), and allows the statements to be connected in loops. Which means that the execution order matters a lot. Which means that there are some quite extensive rules, determining this order. The logic ends up being very much procedural, but written in the peculiar way of SQL statements and connecting streams.

The good thing is that all this allows to control the execution order very closely and write things that are very difficult to express in pure un-ordered relational operators. Which allows to aggregate the data early and creatively, keeping less data in memory.

The bad news is that it limits the execution to a single thread. If you want a separate thread, you must explicitly make a separate module, and program the communications between the modules, which is not exactly easy to get right. There are lots of people who do it the easy way and then wonder, why do they get the occasional data corruption. Also, the ordering rules for execution inside a module are quite tricky. Even for fairly simple logic, it requires writing a lot of code, some of which is just bulky (try enumerating 90 fields in each statement), and some of which is tricky to get right.

The summary is that everything is not what it seems: the Aleri models aren't usually written in SQL but are very declarative in their meaning, while the Coral8/StreamBase models are written in an SQL-like language but in reality are totally procedural.

Sybase is also striking for a middle ground, combining the features inherited from Aleri and Coral8 in its CEP R5 and later: use the CCL language but relax the execution order rules to the Aleri level, except for the explicit single-threaded sections where the order is important. Include the SPLASH fragments for where the outright procedural logic is easy to use. Even though it sounds hodgy-podgy, it actually came together pretty nicely. Forgive me for saying so myself since I've done a fair amount of design and the execution logic implementation for it before I've left Sybase.

Still, not everything is perfect in this merged world. The SQLy syntax still requires you to drag around all your 90 fields into nearly  every statement.  The single-threaded order of execution is still non-obvious. It's possible to write the procedural code directly in SPLASH but the boundary where the data passes between the SQLy and C-ish code still has a whole lot of its own kinks (less than in Aleri). And worst of all, there is still no modular programming. Yeah, there are "modules" but they are not really reusable. They are tied too tightly to the schema of the data. What is needed, is more like C++ templates.