Showing posts with label tql. Show all posts
Showing posts with label tql. Show all posts
Wednesday, April 22, 2015
ThreadedClient improvement of timeout handling
The work on updating the tests for the new ordered index brought a nice side effect: the handling of the timeouts in the Perl class X:ThreadedClient has improved. Now it returns not only the error message but also the data received up to that point. This helps a lot with diagnosing the errors in the automated tests of TQL: the error messages get returned in a clear way.
Tuesday, July 29, 2014
compact vs detailed
A few days ago I've been editing the Developer's Guide chapter on TQL when an idea had occurred to me:
Perhaps, what makes the typical CEP model so hard to read is that they try to represent both the topology and all the details in one go. Yes, I'm a fan of reading the program text sequentially, and I've been saying that the problem with SQL is that it's written backwards: each statement specifies the results first and then the computation; if the results went after computation, the whole flow would be much easier to read sequentially. But it still doesn't solve the problem of branching, and maybe doesn't solve the fundamental difficulty at all.
Maybe the right way to write the CEP programs is to write out the pipelines in a short notation, and then define the elements of these pipelines in all detail separately. So the main program might look something like:
inputTrades | checkTrades -> storeTrades
inputSymbols -> storeSymbols
(storeTrades, storeSymbols) | enrichTrades | outputData
and then each of these stages can be defined in detail separately. No arguments, no details at all in the pipeline specification. All the details would go into the detailed definitions. So each module would consist of these two parts: the pipeline outline and the detailed definitions of the pipeline elements. And then these modules can be combined into the higher-level modules.
Though it's still not clear, how to show say the maintenance with the expiration of the data in this notation? Would it make sense in the more complicated cases?
Perhaps, what makes the typical CEP model so hard to read is that they try to represent both the topology and all the details in one go. Yes, I'm a fan of reading the program text sequentially, and I've been saying that the problem with SQL is that it's written backwards: each statement specifies the results first and then the computation; if the results went after computation, the whole flow would be much easier to read sequentially. But it still doesn't solve the problem of branching, and maybe doesn't solve the fundamental difficulty at all.
Maybe the right way to write the CEP programs is to write out the pipelines in a short notation, and then define the elements of these pipelines in all detail separately. So the main program might look something like:
inputTrades | checkTrades -> storeTrades
inputSymbols -> storeSymbols
(storeTrades, storeSymbols) | enrichTrades | outputData
and then each of these stages can be defined in detail separately. No arguments, no details at all in the pipeline specification. All the details would go into the detailed definitions. So each module would consist of these two parts: the pipeline outline and the detailed definitions of the pipeline elements. And then these modules can be combined into the higher-level modules.
Though it's still not clear, how to show say the maintenance with the expiration of the data in this notation? Would it make sense in the more complicated cases?
Sunday, July 20, 2014
a minor TQL clean-up
I've been working on the documentation chapter about TQL, and noticed that its single-threaded version has not been updated to use the query name as the label in the response. So I've fixed it. The previous output like
query,{read table tSymbol}
lb1read OP_INSERT symbol="AAA" name="Absolute Auto Analytics Inc" eps="0.5"
+EOD,OP_NOP,lb1read
now becomes
query,{read table tSymbol}
query OP_INSERT symbol="AAA" name="Absolute Auto Analytics Inc" eps="0.5"
+EOD,OP_NOP,query
I've also added the optional argument for name override in SimpleServer::makeServerOutLabel().
query,{read table tSymbol}
lb1read OP_INSERT symbol="AAA" name="Absolute Auto Analytics Inc" eps="0.5"
+EOD,OP_NOP,lb1read
now becomes
query,{read table tSymbol}
query OP_INSERT symbol="AAA" name="Absolute Auto Analytics Inc" eps="0.5"
+EOD,OP_NOP,query
I've also added the optional argument for name override in SimpleServer::makeServerOutLabel().
Friday, December 20, 2013
PowerShell
First, a status update: I've finally resumed the work on the docs for 2.0, albeit it's proceeding slowly yet.
I've been reading recently on PowerShell. A pretty cool tool. I've tried it before and it didn't work well for me then because I didn't understand its purpose. It's not a normal OS shell. Instead, it's the shell for the .NET virtual machine. Exactly the thing that Java is missing, and the gap that it tries to plug with the crap like Ant and Maven, unsuccessfully. PowerShell lets you run all the .NET methods interactively from the command line, and build the pipelines of them. It has some very cool syntax that lets you automatically apply the pipeline input in the same way as the command-line input. It also has the remote execution functionality, so it serves as an analog of the rsh/ssh (more advanced in some ways, less advanced in the others) in the Microsoft ecosystem.
But here is the CEP-related part: The Triceps TQL is not unique in handling the SQL-like queries in the form of pipelines. PowerShell does that too. I guess, it's a fairly obvious idea. You can even treat PowerShell as a rudimentary CEP system, and write the processing in the form of CEP pipelines. There is a major limitation that the pipelines are all linear, with no forking and joining, but on the other hand the pipelines can be used to pass the arbitrarily complex objects, and also a mix of objects of different types, so with some creativity the more complex topologies can be simulated (still no loops though).
Another catch with the pipelines is very similar to a current limitation of TQL: each stage of the pipeline works all by itself. In a SQL statement the query optimizer can turn a WHERE clause into an iteration by a small subset of an index. In TQL and PowerShell there will be an iteration on everything, followed by filtering in a WHERE. The grand plan for TQL is to add a query optimizer to it eventually, that could combine multiple sequential stages of the pipeline into one optimized stage. The other, simpler, alternative that I was considering for the short term is to specify the optimized selection manually as an option to the command that reads the table contents. PowerShell takes that one in many cases, so that say the command that pulls the data from a SqlServer database gets a full SQL query as an argument and does its filtering on the server. But I guess theoretically nothing really stops them from doing some pipeline optimization by pulling the WHERE conditions from a "where" command into the SQL statement for the database selection command. If would save the trouble of the SQL and "where" having different syntax.
I've been reading recently on PowerShell. A pretty cool tool. I've tried it before and it didn't work well for me then because I didn't understand its purpose. It's not a normal OS shell. Instead, it's the shell for the .NET virtual machine. Exactly the thing that Java is missing, and the gap that it tries to plug with the crap like Ant and Maven, unsuccessfully. PowerShell lets you run all the .NET methods interactively from the command line, and build the pipelines of them. It has some very cool syntax that lets you automatically apply the pipeline input in the same way as the command-line input. It also has the remote execution functionality, so it serves as an analog of the rsh/ssh (more advanced in some ways, less advanced in the others) in the Microsoft ecosystem.
But here is the CEP-related part: The Triceps TQL is not unique in handling the SQL-like queries in the form of pipelines. PowerShell does that too. I guess, it's a fairly obvious idea. You can even treat PowerShell as a rudimentary CEP system, and write the processing in the form of CEP pipelines. There is a major limitation that the pipelines are all linear, with no forking and joining, but on the other hand the pipelines can be used to pass the arbitrarily complex objects, and also a mix of objects of different types, so with some creativity the more complex topologies can be simulated (still no loops though).
Another catch with the pipelines is very similar to a current limitation of TQL: each stage of the pipeline works all by itself. In a SQL statement the query optimizer can turn a WHERE clause into an iteration by a small subset of an index. In TQL and PowerShell there will be an iteration on everything, followed by filtering in a WHERE. The grand plan for TQL is to add a query optimizer to it eventually, that could combine multiple sequential stages of the pipeline into one optimized stage. The other, simpler, alternative that I was considering for the short term is to specify the optimized selection manually as an option to the command that reads the table contents. PowerShell takes that one in many cases, so that say the command that pulls the data from a SqlServer database gets a full SQL query as an argument and does its filtering on the server. But I guess theoretically nothing really stops them from doing some pipeline optimization by pulling the WHERE conditions from a "where" command into the SQL statement for the database selection command. If would save the trouble of the SQL and "where" having different syntax.
Monday, May 27, 2013
how to export a table, or the guts of TQL join exposed
Now to the point of why the multithreaded TQl example got written: the export of a table between two threads.
It all starts in the Tql initialization method. In the multithreaded mode it builds the nexuses for communication. I'll skip the input nexus and show the building of only the output and request-dump nexuses:
# row type for dump requests and responses
my $rtRequest = Triceps::RowType->new(
client => "string", #requesting client
id => "string", # request id
name => "string", # the table name, for convenience of requestor
cmd => "string", # for convenience of requestor, the command that it is executing
) or confess "$!";
The request row type is used by the client writer thread to request the table dumps from the core logic, and to get back the notifications about the dumps.
# build the output side
for (my $i = 0; $i <= $#{$self->{tables}}; $i++) {
my $name = $self->{tableNames}[$i];
my $table = $self->{tables}[$i];
push @tabtypes, $name, $table->getType()->copyFundamental();
push @labels, "t.out." . $name, $table->getOutputLabel();
push @labels, "t.dump." . $name, $table->getDumpLabel();
}
push @labels, "control", $rtControl; # pass-through from in to out
push @labels, "beginDump", $rtRequest; # framing for the table dumps
push @labels, "endDump", $rtRequest;
$self->{faOut} = $owner->makeNexus(
name => $self->{nxprefix} . "out",
labels => [ @labels ],
tableTypes => [ @tabtypes ],
import => "writer",
);
$self->{beginDump} = $self->{faOut}->getLabel("beginDump");
$self->{endDump} = $self->{faOut}->getLabel("endDump");
On the output side each table is represented by 3 elements:
The row type $rtControl and label "control" is used to pass the commands from the client reader to client writer, but it's exact contents is not important here.
The dump request nexus is built in a similar way:
# build the dump requests, will be coming from below
undef @labels;
for (my $i = 0; $i <= $#{$self->{tables}}; $i++) {
my $name = $self->{tableNames}[$i];
my $table = $self->{tables}[$i];
push @labels, "t.rqdump." . $name, $rtRequest;
}
$self->{faRqDump} = $owner->makeNexus(
name => $self->{nxprefix} . "rqdump",
labels => [ @labels ],
reverse => 1, # avoids making a loop, and gives priority
import => "reader",
);
# tie together the labels
for (my $i = 0; $i <= $#{$self->{tables}}; $i++) {
my $name = $self->{tableNames}[$i];
my $table = $self->{tables}[$i];
$self->{faRqDump}->getLabel("t.rqdump." . $name)->makeChained(
$self->{nxprefix} . "rqdump." . $name, undef,
\&_dumpTable, $self, $table
);
}
The dumps are executed in the function _dumpTable:
sub _dumpTable # ($label, $rowop, $self, $table)
{
my ($label, $rop, $self, $table) = @_;
my $unit = $label->getUnit();
# pass through the client id to the dump
$unit->call($self->{beginDump}->adopt($rop));
$table->dumpAll();
$unit->call($self->{endDump}->adopt($rop));
$self->{faOut}->flushWriter();
}
The data gets framed around by the "beginDump" and "endDump" labels getting the copies of the original request. This helps the client writer thread keep track of its current spot. The flushing of the writer is not strictly needed. Just in case if multiple dump requests are received in a single tray, it breaks up the responses into a separate tray for each dump, keeping the size of the trays lower. Not that this situation could actually happen yet.
This part taken care of, let's jump around and see how the client writer thread processes a "querysub" request:
} elsif ($cmd eq "querysub") {
if ($id eq "" || exists $queries{$id}) {
printOrShut($app, $fragment, $sock,
"error,$id,Duplicate id '$id': query ids must be unique,bad_id,$id\n");
next;
}
my $ctx = compileQuery(
qid => $id,
qname => $args[0],
text => $args[1],
subError => sub {
chomp $_[2];
$_[2] =~ s/\n/\\n/g; # no real newlines in the output
$_[2] =~ s/,/;/g; # no confusing commas in the output
printOrShut($app, $fragment, $sock, "error,", join(',', @_), "\n");
},
faOut => $faOut,
faRqDump => $faRqDump,
subPrint => sub {
printOrShut($app, $fragment, $sock, @_);
},
);
if ($ctx) { # otherwise the error is already reported
$queries{$id} = $ctx;
&$runNextRequest($ctx);
}
}
The query id is used to keep track of the outstanding queries, so the code makes sure that it's unique, and you can see an example of the query response. The bulk of the work is done in the method compileQuery(). The arguments to it give the details of the query and also provide the closures for the functionality that differs between the single-threaded and multi-threaded versions. The option "subError" is used to send the errors to the client, and "subPrint" is used to send the output to the client, it gets used for building the labels in the "print" command of the query.
compileQuery() returns the query context, which contains a compiled sub-model that executes the query and a set of requests that tell the writer how to connect the query to the incoming data. Or on error it reports the error using subError and returns an undef. If the compilation succeeded, the writer remembers the query and starts the asynchronous execution of the requests. More about the requests later, now let's look at the query compilation and context.
The context is created in compileQuery() thusly:
my $ctx = {};
$ctx->{qid} = $opts->{qid};
$ctx->{qname} = $opts->{qname};
# .. skipped the parts related to single-threadde TQL
$ctx->{faOut} = $opts->{faOut};
$ctx->{faRqDump} = $opts->{faRqDump};
$ctx->{subPrint} = $opts->{subPrint};
$ctx->{requests} = []; # dump and subscribe requests that will run the pipeline
$ctx->{copyTables} = []; # the tables created in this query
# (have to keep references to the tables or they will disappear)
# The query will be built in a separate unit
$ctx->{u} = Triceps::Unit->new($opts->{nxprefix} . "${q}.unit");
$ctx->{prev} = undef; # will contain the output of the previous command in the pipeline
$ctx->{id} = 0; # a unique id for auto-generated objects
# deletion of the context will cause the unit in it to clean
$ctx->{cleaner} = $ctx->{u}->makeClearingTrigger();
It has some parts common and some parts differing for the single- and multi-threaded varieties, here I've skipped over the single-threaded parts.
One element that is left undefined here is $ctx->{prev}. It's the label created as the output of the previous stage of the query pipeline. As each command in the pipeline builds its piece of processing, it chains its logic from $ctx->{prev} and leaves its result label in $ctx->{next}. Then compileQuery() moves "next" to "prev" and calls the compilation of the next command in the pipeline. The only command that accepts an undefined "prev" (and it must be undefined for it) is "read", that reads the table at the start of the pipeline.
$ctx->{copyTables} also has an important point behind it. When you create a label, it's OK to discard the original reference after you chain the label into the logic, that chaining will keep a reference and the label will stay alive. Not so with a table: if you create a table, chain its input label and then drop the reference to a table, the table will be discarded. Then when the input label will try to send any data to the table, it will die (and unless very recently it outright crashed). So it's important to keep the table reference alive, and that's what this array is for.
$ctx->{id} is used to generate the unique names for the objects build in a query.
Each query is built in its own unit. This is convenient, after the query is done or the compilation encounters an error, the unit with its whole contents gets easily discarded. The clearing trigger placed in the context makes sure that the unit gets properly cleared and discarded.
Next goes the compilation of the join query command, I'll go through it in chunks.
sub _tqlJoin # ($ctx, @args)
{
my $ctx = shift;
die "The join command may not be used at the start of a pipeline.\n"
unless (defined($ctx->{prev}));
my $opts = {};
&Triceps::Opt::parse("join", $opts, {
table => [ undef, \&Triceps::Opt::ck_mandatory ],
rightIdxPath => [ undef, undef ],
by => [ undef, undef ],
byLeft => [ undef, undef ],
leftFields => [ undef, undef ],
rightFields => [ undef, undef ],
type => [ "inner", undef ],
}, @_);
my $tabname = bunquote($opts->{table});
my $unit = $ctx->{u};
my $table;
&Triceps::Opt::checkMutuallyExclusive("join", 1, "by", $opts->{by}, "byLeft", $opts->{byLeft});
my $by = split_braced_final($opts->{by});
my $byLeft = split_braced_final($opts->{byLeft});
my $rightIdxPath;
if (defined $opts->{rightIdxPath}) { # propagate the undef
$rightIdxPath = split_braced_final($opts->{rightIdxPath});
}
It starts by parsing the options and converting them to the internal representation, removing the braced quotes.
if ($ctx->{faOut}) {
# Potentially, the tables might be reused between multiple joins
# in the query if the required keys match. But for now keep things
# simpler by creating a new table from scratch each time.
my $tt = eval {
# copy to avoid adding an index to the original type
$ctx->{faOut}->impTableType($tabname)->copy();
};
die ("Join found no such table '$tabname'\n") unless ($tt);
if (!defined $rightIdxPath) {
# determine or add the index automatically
my @workby;
if (defined $byLeft) { # need to translate
my @leftfld = $ctx->{prev}->getRowType()->getFieldNames();
@workby = &Triceps::Fields::filterToPairs("Join option 'byLeft'",
\@leftfld, [ @$byLeft, "!.*" ]);
} else {
@workby = @$by;
}
my @idxkeys; # extract the keys for the right side table
for (my $i = 1; $i <= $#workby; $i+= 2) {
push @idxkeys, $workby[$i];
}
$rightIdxPath = [ $tt->findOrAddIndex(@idxkeys) ];
}
# build the table from the type
$tt->initialize() or confess "$!";
$table = $ctx->{u}->makeTable($tt, "EM_CALL", "tab" . $ctx->{id} . $tabname);
push @{$ctx->{copyTables}}, $table;
# build the request that fills the table with data and then
# keeps it up to date;
# the table has to be filled before the query's main flow starts,
# so put the request at the front
&_makeQdumpsub($ctx, $tabname, 1, $table->getInputLabel());
} else {
die ("Join found no such table '$tabname'\n")
unless (exists $ctx->{tables}{$tabname});
$table = $ctx->{tables}{$tabname};
}
The presence of $ctx->{faOut} means that the query is compiled in the multithreaded context.
The command handles may freely die, and the error messages will be caught by compileQuery() and nicely (at least, sort-of) reported back to the user.
If an explicit rightIdxPath was not requested, it gets found or added automatically. On the way there the index fields need to be determined. Which can be specified as either explicit pairs in the option "by" or the in the name translation syntax in the option "byLeft". If we've got a "byLeft", first it gets translated to the same format as "by", and then the right-side fields are extracted from the format of "by". After that $tt->findOrAddIndex() takes care of all the heavy lifting. It either finds a matching index type in the table type or creates a new one from the specified fields, and either way returns the index path. (An invalid field will make it confess).
It looks a bit anti-climactic, but the three lines of exporting with copyFundamental(), impTableType() and findOrAddIndex() is what this large example is all about.
You might wonder, how come the explicit rightIdxPath is not checked in any way? It will be checked later by LookupJoin(), so not much point in doing the check twice.
After that the table is created in a straightforward way, and rememebered in copyTables. And the requests list gets prepended with a request to dump and subscribe to this table. I'll get back to that, for now let's finish up with _tqlJoin().
my $isLeft = 0; # default for inner join
my $type = $opts->{type};
if ($type eq "inner") {
# already default
} elsif ($type eq "left") {
$isLeft = 1;
} else {
die "Unsupported value '$type' of option 'type'.\n"
}
my $leftFields = split_braced_final($opts->{leftFields});
my $rightFields = split_braced_final($opts->{rightFields});
my $join = Triceps::LookupJoin->new(
name => "join" . $ctx->{id},
unit => $unit,
leftFromLabel => $ctx->{prev},
rightTable => $table,
rightIdxPath => $rightIdxPath,
leftFields => $leftFields,
rightFields => $rightFields,
by => $by,
byLeft => $byLeft,
isLeft => $isLeft,
fieldsDropRightKey => 1,
);
$ctx->{next} = $join->getOutputLabel();
}
The rest of the options get parsed, and then all the collected data gets forwarded to the LookupJoin constructor. Finally the "next" label is assigned from the join's result.
Now jumping to the _makeQdumpsub(). It's used by both the "read" and "join" query commands to initiate the joins and subscriptions.
sub _makeQdumpsub # ($ctx, $tabname, [$front, $lbNext])
{
my $ctx = shift;
my $tabname = shift;
my $front = shift;
my $lbNext = shift;
my $unit = $ctx->{u};
my $lbrq = eval {
$ctx->{faRqDump}->getLabel("t.rqdump.$tabname");
};
my $lbsrc = eval {
$ctx->{faOut}->getLabel("t.out.$tabname");
};
die ("Found no such table '$tabname'\n") unless ($lbrq && $lbsrc);
# compute the binding for the data dumps, that would be a cross-unit
# binding to the original faOut but it's OK
my $fretOut = $ctx->{faOut}->getFnReturn();
my $dumpname = "t.dump.$tabname";
# the dump and following subscription data will merge on this label
if (!defined $lbNext) {
$lbNext = $unit->makeDummyLabel(
$lbsrc->getRowType(), "lb" . $ctx->{id} . "out_$tabname");
}
my $bindDump = Triceps::FnBinding->new(
on => $fretOut,
name => "bind" . $ctx->{id} . "dump",
labels => [ $dumpname => $lbNext ],
);
First it finds all the proper labels. The label $lbNext will accept the merged dump contents and the following subscription, and it might be either auto-generated or received as an argument. A join pass it as an argument, $table->getInputLabel(), so all the data goes to the copied table.
The binding is used to receive the dump. It's a bit of an optimization. Remember, the dump labels are shared between all the clients. Whenever any client requests a dump, all the clients will get the response. A client finds that the incoming dump is destined for it by processing the "beginDump" label. If it contains this client's name, the dump is destined here, and the client reacts by pushing the appropriate binding onto the facet's FnReturn, and the data flows. The matching "endDump" label then pops the binding and the data stops flowing. The binding allows to avoid checking every rowop for whethere it's supposed to be accepted and if yes then where exactly (rememeber, the same table may be dumped independently multiple times by multiple queries). Just check once at the start of the bundle and then let the data flow in bulk.
# qdumpsub:
# * label where to send the dump request to
# * source output label, from which a subscription will be set up
# at the end of the dump
# * target label in the query that will be tied to the source label
# * binding to be used during the dump, which also directs the data
# to the same target label
my $request = [ "qdumpsub", $lbrq, $lbsrc, $lbNext, $bindDump ];
if ($front) {
unshift @{$ctx->{requests}}, $request;
} else {
push @{$ctx->{requests}}, $request;
}
return $lbNext;
}
Finally, the created bits and pieces get packaged into a request and added to the list of requests in the query context. The last tricky part is that the request can be added at the back or the front of the list. The "normal" way is to add to the back, however the dimension tables for the joins have to be populated before the main data flow of the query starts. So for them the argument $front is set to 1, and they get added in the front.
Now jumping back to the writer thread logic, after it called compileQuery, it starts the query execution by calling &$runNextRequest(). Which is a closure function defined inside the client writer function, and knows how to process the "qdumpsub"s we've just seen created.
my $runNextRequest = sub { # ($ctx)
my $ctx = shift;
my $requests = $ctx->{requests};
undef $ctx->{curRequest}; # clear the info of the previous request
my $r = shift @$requests;
if (!defined $r) {
# all done, now just need to pump the data through
printOrShut($app, $fragment, $sock,
"querysub,$ctx->{qid},$ctx->{qname}\n");
return;
}
First it clears the information about the previous request, if any. This function will be called after each request, to send on the next one, so on all its calls except the first one for a query it will have something to clear.
Then it checks if all the requests are already done. If so, it sends the query confirmation to the client and returns. The subscription part of the query will continue running on its own.
$ctx->{curRequest} = $r; # remember until completed
my $cmd = $$r[0];
if ($cmd eq "qdumpsub") {
# qdumpsub:
# * label where to send the dump request to
# * source output label, from which a subscription will be set up
# at the end of the dump
# * target label in the query that will be tied to the source label
# * binding to be used during the dump, which also directs the data
# to the same target label
my $lbrq = $$r[1];
$unit->makeHashCall($lbrq, "OP_INSERT",
client => $fragment, id => $ctx->{qid}, name => $ctx->{qname}, cmd => $cmd);
The "qdumpsub" gets forwarded to the core logic. The responses will be processed in the handlers or "beginDump" and "endDump". One of the great pains of this "actor" architecture is that the linear logic gets broken up into many disjointed pieces in the separate handlers.
} else {
printOrShut($app, $fragment, $sock,
"error,", $ctx->{qid}, ",Internal error: unknown request '$cmd',internal,", $cmd, "\n");
$ctx->{requests} = [];
undef $ctx->{curRequest};
# and this will leave the query partially initialized,
# but it should never happen
return;
}
};
And a catch-all just in case if the query compiler ever decides to produce an invalid request.
Next goes the handling of the dump labels (again, this gets set up during the build of the client reader threads, and then the nature is left to run its course, reacting to the rowops as they come in).
$faOut->getLabel("beginDump")->makeChained("lbBeginDump", undef, sub {
my $row = $_[1]->getRow();
my ($client, $id, $name, $cmd) = $row->toArray();
return unless ($client eq $fragment);
if ($cmd eq "qdumpsub") {
return unless(exists $queries{$id});
my $ctx = $queries{$id};
$fretOut->push($ctx->{curRequest}[4]); # the binding for the dump
} else {
# .. skipped the handling of dump/dumpsub
}
});
All it does is checks if this is the destination client, and if there is an active request with this id, then it pushes the appropriate binding.
$faOut->getLabel("endDump")->makeChained("lbEndDump", undef, sub {
my $row = $_[1]->getRow();
my ($client, $id, $name, $cmd) = $row->toArray();
return unless ($client eq $fragment);
if ($cmd eq "qdumpsub") {
return unless(exists $queries{$id});
my $ctx = $queries{$id};
$fretOut->pop($ctx->{curRequest}[4]); # the binding for the dump
# and chain together all the following updates
$ctx->{curRequest}[2]->makeChained(
"qsub$id." . $ctx->{curRequest}[3]->getName(), undef,
sub {
# a cross-unit call
$_[2]->call($_[3]->adopt($_[1]));
},
$ctx->{u}, $ctx->{curRequest}[3]
);
&$runNextRequest($ctx);
} else {
# .. skipped the handling of dump/dumpsub
}
});
Same things as the "beginDump", checks if this is the right client, and if it has an outstanding dump request, then pops the binding. After the dump is completed, the subscription has to be set up, so it sets up a label that forwards the normal output of this table to the label specified in the request. Since each query is defined in its own unit, this forwarding is done as a cross-unit call.
And then the next request of this query can be started.
By the way, the cross-unit adopt() didn't work in Perl until I wrote this example. There was a check against it (the C++ API never bothered with this check). But the adoption between the units has turned out to be quite convenient, so I've removed that check.
And that's it. Long and winding but finally completed. It's kind of about only three lines of code, but I think the rest of it also shows the useful techniques of the work with threads.
It all starts in the Tql initialization method. In the multithreaded mode it builds the nexuses for communication. I'll skip the input nexus and show the building of only the output and request-dump nexuses:
# row type for dump requests and responses
my $rtRequest = Triceps::RowType->new(
client => "string", #requesting client
id => "string", # request id
name => "string", # the table name, for convenience of requestor
cmd => "string", # for convenience of requestor, the command that it is executing
) or confess "$!";
The request row type is used by the client writer thread to request the table dumps from the core logic, and to get back the notifications about the dumps.
# build the output side
for (my $i = 0; $i <= $#{$self->{tables}}; $i++) {
my $name = $self->{tableNames}[$i];
my $table = $self->{tables}[$i];
push @tabtypes, $name, $table->getType()->copyFundamental();
push @labels, "t.out." . $name, $table->getOutputLabel();
push @labels, "t.dump." . $name, $table->getDumpLabel();
}
push @labels, "control", $rtControl; # pass-through from in to out
push @labels, "beginDump", $rtRequest; # framing for the table dumps
push @labels, "endDump", $rtRequest;
$self->{faOut} = $owner->makeNexus(
name => $self->{nxprefix} . "out",
labels => [ @labels ],
tableTypes => [ @tabtypes ],
import => "writer",
);
$self->{beginDump} = $self->{faOut}->getLabel("beginDump");
$self->{endDump} = $self->{faOut}->getLabel("endDump");
On the output side each table is represented by 3 elements:
- its fundamental table type (stripped down to the primary key);
- its output label for normal updates;
- its dump label for the responses to the dump requests.
The row type $rtControl and label "control" is used to pass the commands from the client reader to client writer, but it's exact contents is not important here.
The dump request nexus is built in a similar way:
# build the dump requests, will be coming from below
undef @labels;
for (my $i = 0; $i <= $#{$self->{tables}}; $i++) {
my $name = $self->{tableNames}[$i];
my $table = $self->{tables}[$i];
push @labels, "t.rqdump." . $name, $rtRequest;
}
$self->{faRqDump} = $owner->makeNexus(
name => $self->{nxprefix} . "rqdump",
labels => [ @labels ],
reverse => 1, # avoids making a loop, and gives priority
import => "reader",
);
# tie together the labels
for (my $i = 0; $i <= $#{$self->{tables}}; $i++) {
my $name = $self->{tableNames}[$i];
my $table = $self->{tables}[$i];
$self->{faRqDump}->getLabel("t.rqdump." . $name)->makeChained(
$self->{nxprefix} . "rqdump." . $name, undef,
\&_dumpTable, $self, $table
);
}
The dumps are executed in the function _dumpTable:
sub _dumpTable # ($label, $rowop, $self, $table)
{
my ($label, $rop, $self, $table) = @_;
my $unit = $label->getUnit();
# pass through the client id to the dump
$unit->call($self->{beginDump}->adopt($rop));
$table->dumpAll();
$unit->call($self->{endDump}->adopt($rop));
$self->{faOut}->flushWriter();
}
The data gets framed around by the "beginDump" and "endDump" labels getting the copies of the original request. This helps the client writer thread keep track of its current spot. The flushing of the writer is not strictly needed. Just in case if multiple dump requests are received in a single tray, it breaks up the responses into a separate tray for each dump, keeping the size of the trays lower. Not that this situation could actually happen yet.
This part taken care of, let's jump around and see how the client writer thread processes a "querysub" request:
} elsif ($cmd eq "querysub") {
if ($id eq "" || exists $queries{$id}) {
printOrShut($app, $fragment, $sock,
"error,$id,Duplicate id '$id': query ids must be unique,bad_id,$id\n");
next;
}
my $ctx = compileQuery(
qid => $id,
qname => $args[0],
text => $args[1],
subError => sub {
chomp $_[2];
$_[2] =~ s/\n/\\n/g; # no real newlines in the output
$_[2] =~ s/,/;/g; # no confusing commas in the output
printOrShut($app, $fragment, $sock, "error,", join(',', @_), "\n");
},
faOut => $faOut,
faRqDump => $faRqDump,
subPrint => sub {
printOrShut($app, $fragment, $sock, @_);
},
);
if ($ctx) { # otherwise the error is already reported
$queries{$id} = $ctx;
&$runNextRequest($ctx);
}
}
The query id is used to keep track of the outstanding queries, so the code makes sure that it's unique, and you can see an example of the query response. The bulk of the work is done in the method compileQuery(). The arguments to it give the details of the query and also provide the closures for the functionality that differs between the single-threaded and multi-threaded versions. The option "subError" is used to send the errors to the client, and "subPrint" is used to send the output to the client, it gets used for building the labels in the "print" command of the query.
compileQuery() returns the query context, which contains a compiled sub-model that executes the query and a set of requests that tell the writer how to connect the query to the incoming data. Or on error it reports the error using subError and returns an undef. If the compilation succeeded, the writer remembers the query and starts the asynchronous execution of the requests. More about the requests later, now let's look at the query compilation and context.
The context is created in compileQuery() thusly:
my $ctx = {};
$ctx->{qid} = $opts->{qid};
$ctx->{qname} = $opts->{qname};
# .. skipped the parts related to single-threadde TQL
$ctx->{faOut} = $opts->{faOut};
$ctx->{faRqDump} = $opts->{faRqDump};
$ctx->{subPrint} = $opts->{subPrint};
$ctx->{requests} = []; # dump and subscribe requests that will run the pipeline
$ctx->{copyTables} = []; # the tables created in this query
# (have to keep references to the tables or they will disappear)
# The query will be built in a separate unit
$ctx->{u} = Triceps::Unit->new($opts->{nxprefix} . "${q}.unit");
$ctx->{prev} = undef; # will contain the output of the previous command in the pipeline
$ctx->{id} = 0; # a unique id for auto-generated objects
# deletion of the context will cause the unit in it to clean
$ctx->{cleaner} = $ctx->{u}->makeClearingTrigger();
It has some parts common and some parts differing for the single- and multi-threaded varieties, here I've skipped over the single-threaded parts.
One element that is left undefined here is $ctx->{prev}. It's the label created as the output of the previous stage of the query pipeline. As each command in the pipeline builds its piece of processing, it chains its logic from $ctx->{prev} and leaves its result label in $ctx->{next}. Then compileQuery() moves "next" to "prev" and calls the compilation of the next command in the pipeline. The only command that accepts an undefined "prev" (and it must be undefined for it) is "read", that reads the table at the start of the pipeline.
$ctx->{copyTables} also has an important point behind it. When you create a label, it's OK to discard the original reference after you chain the label into the logic, that chaining will keep a reference and the label will stay alive. Not so with a table: if you create a table, chain its input label and then drop the reference to a table, the table will be discarded. Then when the input label will try to send any data to the table, it will die (and unless very recently it outright crashed). So it's important to keep the table reference alive, and that's what this array is for.
$ctx->{id} is used to generate the unique names for the objects build in a query.
Each query is built in its own unit. This is convenient, after the query is done or the compilation encounters an error, the unit with its whole contents gets easily discarded. The clearing trigger placed in the context makes sure that the unit gets properly cleared and discarded.
Next goes the compilation of the join query command, I'll go through it in chunks.
sub _tqlJoin # ($ctx, @args)
{
my $ctx = shift;
die "The join command may not be used at the start of a pipeline.\n"
unless (defined($ctx->{prev}));
my $opts = {};
&Triceps::Opt::parse("join", $opts, {
table => [ undef, \&Triceps::Opt::ck_mandatory ],
rightIdxPath => [ undef, undef ],
by => [ undef, undef ],
byLeft => [ undef, undef ],
leftFields => [ undef, undef ],
rightFields => [ undef, undef ],
type => [ "inner", undef ],
}, @_);
my $tabname = bunquote($opts->{table});
my $unit = $ctx->{u};
my $table;
&Triceps::Opt::checkMutuallyExclusive("join", 1, "by", $opts->{by}, "byLeft", $opts->{byLeft});
my $by = split_braced_final($opts->{by});
my $byLeft = split_braced_final($opts->{byLeft});
my $rightIdxPath;
if (defined $opts->{rightIdxPath}) { # propagate the undef
$rightIdxPath = split_braced_final($opts->{rightIdxPath});
}
It starts by parsing the options and converting them to the internal representation, removing the braced quotes.
if ($ctx->{faOut}) {
# Potentially, the tables might be reused between multiple joins
# in the query if the required keys match. But for now keep things
# simpler by creating a new table from scratch each time.
my $tt = eval {
# copy to avoid adding an index to the original type
$ctx->{faOut}->impTableType($tabname)->copy();
};
die ("Join found no such table '$tabname'\n") unless ($tt);
if (!defined $rightIdxPath) {
# determine or add the index automatically
my @workby;
if (defined $byLeft) { # need to translate
my @leftfld = $ctx->{prev}->getRowType()->getFieldNames();
@workby = &Triceps::Fields::filterToPairs("Join option 'byLeft'",
\@leftfld, [ @$byLeft, "!.*" ]);
} else {
@workby = @$by;
}
my @idxkeys; # extract the keys for the right side table
for (my $i = 1; $i <= $#workby; $i+= 2) {
push @idxkeys, $workby[$i];
}
$rightIdxPath = [ $tt->findOrAddIndex(@idxkeys) ];
}
# build the table from the type
$tt->initialize() or confess "$!";
$table = $ctx->{u}->makeTable($tt, "EM_CALL", "tab" . $ctx->{id} . $tabname);
push @{$ctx->{copyTables}}, $table;
# build the request that fills the table with data and then
# keeps it up to date;
# the table has to be filled before the query's main flow starts,
# so put the request at the front
&_makeQdumpsub($ctx, $tabname, 1, $table->getInputLabel());
} else {
die ("Join found no such table '$tabname'\n")
unless (exists $ctx->{tables}{$tabname});
$table = $ctx->{tables}{$tabname};
}
The presence of $ctx->{faOut} means that the query is compiled in the multithreaded context.
The command handles may freely die, and the error messages will be caught by compileQuery() and nicely (at least, sort-of) reported back to the user.
If an explicit rightIdxPath was not requested, it gets found or added automatically. On the way there the index fields need to be determined. Which can be specified as either explicit pairs in the option "by" or the in the name translation syntax in the option "byLeft". If we've got a "byLeft", first it gets translated to the same format as "by", and then the right-side fields are extracted from the format of "by". After that $tt->findOrAddIndex() takes care of all the heavy lifting. It either finds a matching index type in the table type or creates a new one from the specified fields, and either way returns the index path. (An invalid field will make it confess).
It looks a bit anti-climactic, but the three lines of exporting with copyFundamental(), impTableType() and findOrAddIndex() is what this large example is all about.
You might wonder, how come the explicit rightIdxPath is not checked in any way? It will be checked later by LookupJoin(), so not much point in doing the check twice.
After that the table is created in a straightforward way, and rememebered in copyTables. And the requests list gets prepended with a request to dump and subscribe to this table. I'll get back to that, for now let's finish up with _tqlJoin().
my $isLeft = 0; # default for inner join
my $type = $opts->{type};
if ($type eq "inner") {
# already default
} elsif ($type eq "left") {
$isLeft = 1;
} else {
die "Unsupported value '$type' of option 'type'.\n"
}
my $leftFields = split_braced_final($opts->{leftFields});
my $rightFields = split_braced_final($opts->{rightFields});
my $join = Triceps::LookupJoin->new(
name => "join" . $ctx->{id},
unit => $unit,
leftFromLabel => $ctx->{prev},
rightTable => $table,
rightIdxPath => $rightIdxPath,
leftFields => $leftFields,
rightFields => $rightFields,
by => $by,
byLeft => $byLeft,
isLeft => $isLeft,
fieldsDropRightKey => 1,
);
$ctx->{next} = $join->getOutputLabel();
}
The rest of the options get parsed, and then all the collected data gets forwarded to the LookupJoin constructor. Finally the "next" label is assigned from the join's result.
Now jumping to the _makeQdumpsub(). It's used by both the "read" and "join" query commands to initiate the joins and subscriptions.
sub _makeQdumpsub # ($ctx, $tabname, [$front, $lbNext])
{
my $ctx = shift;
my $tabname = shift;
my $front = shift;
my $lbNext = shift;
my $unit = $ctx->{u};
my $lbrq = eval {
$ctx->{faRqDump}->getLabel("t.rqdump.$tabname");
};
my $lbsrc = eval {
$ctx->{faOut}->getLabel("t.out.$tabname");
};
die ("Found no such table '$tabname'\n") unless ($lbrq && $lbsrc);
# compute the binding for the data dumps, that would be a cross-unit
# binding to the original faOut but it's OK
my $fretOut = $ctx->{faOut}->getFnReturn();
my $dumpname = "t.dump.$tabname";
# the dump and following subscription data will merge on this label
if (!defined $lbNext) {
$lbNext = $unit->makeDummyLabel(
$lbsrc->getRowType(), "lb" . $ctx->{id} . "out_$tabname");
}
my $bindDump = Triceps::FnBinding->new(
on => $fretOut,
name => "bind" . $ctx->{id} . "dump",
labels => [ $dumpname => $lbNext ],
);
First it finds all the proper labels. The label $lbNext will accept the merged dump contents and the following subscription, and it might be either auto-generated or received as an argument. A join pass it as an argument, $table->getInputLabel(), so all the data goes to the copied table.
The binding is used to receive the dump. It's a bit of an optimization. Remember, the dump labels are shared between all the clients. Whenever any client requests a dump, all the clients will get the response. A client finds that the incoming dump is destined for it by processing the "beginDump" label. If it contains this client's name, the dump is destined here, and the client reacts by pushing the appropriate binding onto the facet's FnReturn, and the data flows. The matching "endDump" label then pops the binding and the data stops flowing. The binding allows to avoid checking every rowop for whethere it's supposed to be accepted and if yes then where exactly (rememeber, the same table may be dumped independently multiple times by multiple queries). Just check once at the start of the bundle and then let the data flow in bulk.
# qdumpsub:
# * label where to send the dump request to
# * source output label, from which a subscription will be set up
# at the end of the dump
# * target label in the query that will be tied to the source label
# * binding to be used during the dump, which also directs the data
# to the same target label
my $request = [ "qdumpsub", $lbrq, $lbsrc, $lbNext, $bindDump ];
if ($front) {
unshift @{$ctx->{requests}}, $request;
} else {
push @{$ctx->{requests}}, $request;
}
return $lbNext;
}
Finally, the created bits and pieces get packaged into a request and added to the list of requests in the query context. The last tricky part is that the request can be added at the back or the front of the list. The "normal" way is to add to the back, however the dimension tables for the joins have to be populated before the main data flow of the query starts. So for them the argument $front is set to 1, and they get added in the front.
Now jumping back to the writer thread logic, after it called compileQuery, it starts the query execution by calling &$runNextRequest(). Which is a closure function defined inside the client writer function, and knows how to process the "qdumpsub"s we've just seen created.
my $runNextRequest = sub { # ($ctx)
my $ctx = shift;
my $requests = $ctx->{requests};
undef $ctx->{curRequest}; # clear the info of the previous request
my $r = shift @$requests;
if (!defined $r) {
# all done, now just need to pump the data through
printOrShut($app, $fragment, $sock,
"querysub,$ctx->{qid},$ctx->{qname}\n");
return;
}
First it clears the information about the previous request, if any. This function will be called after each request, to send on the next one, so on all its calls except the first one for a query it will have something to clear.
Then it checks if all the requests are already done. If so, it sends the query confirmation to the client and returns. The subscription part of the query will continue running on its own.
$ctx->{curRequest} = $r; # remember until completed
my $cmd = $$r[0];
if ($cmd eq "qdumpsub") {
# qdumpsub:
# * label where to send the dump request to
# * source output label, from which a subscription will be set up
# at the end of the dump
# * target label in the query that will be tied to the source label
# * binding to be used during the dump, which also directs the data
# to the same target label
my $lbrq = $$r[1];
$unit->makeHashCall($lbrq, "OP_INSERT",
client => $fragment, id => $ctx->{qid}, name => $ctx->{qname}, cmd => $cmd);
The "qdumpsub" gets forwarded to the core logic. The responses will be processed in the handlers or "beginDump" and "endDump". One of the great pains of this "actor" architecture is that the linear logic gets broken up into many disjointed pieces in the separate handlers.
} else {
printOrShut($app, $fragment, $sock,
"error,", $ctx->{qid}, ",Internal error: unknown request '$cmd',internal,", $cmd, "\n");
$ctx->{requests} = [];
undef $ctx->{curRequest};
# and this will leave the query partially initialized,
# but it should never happen
return;
}
};
And a catch-all just in case if the query compiler ever decides to produce an invalid request.
Next goes the handling of the dump labels (again, this gets set up during the build of the client reader threads, and then the nature is left to run its course, reacting to the rowops as they come in).
$faOut->getLabel("beginDump")->makeChained("lbBeginDump", undef, sub {
my $row = $_[1]->getRow();
my ($client, $id, $name, $cmd) = $row->toArray();
return unless ($client eq $fragment);
if ($cmd eq "qdumpsub") {
return unless(exists $queries{$id});
my $ctx = $queries{$id};
$fretOut->push($ctx->{curRequest}[4]); # the binding for the dump
} else {
# .. skipped the handling of dump/dumpsub
}
});
All it does is checks if this is the destination client, and if there is an active request with this id, then it pushes the appropriate binding.
$faOut->getLabel("endDump")->makeChained("lbEndDump", undef, sub {
my $row = $_[1]->getRow();
my ($client, $id, $name, $cmd) = $row->toArray();
return unless ($client eq $fragment);
if ($cmd eq "qdumpsub") {
return unless(exists $queries{$id});
my $ctx = $queries{$id};
$fretOut->pop($ctx->{curRequest}[4]); # the binding for the dump
# and chain together all the following updates
$ctx->{curRequest}[2]->makeChained(
"qsub$id." . $ctx->{curRequest}[3]->getName(), undef,
sub {
# a cross-unit call
$_[2]->call($_[3]->adopt($_[1]));
},
$ctx->{u}, $ctx->{curRequest}[3]
);
&$runNextRequest($ctx);
} else {
# .. skipped the handling of dump/dumpsub
}
});
Same things as the "beginDump", checks if this is the right client, and if it has an outstanding dump request, then pops the binding. After the dump is completed, the subscription has to be set up, so it sets up a label that forwards the normal output of this table to the label specified in the request. Since each query is defined in its own unit, this forwarding is done as a cross-unit call.
And then the next request of this query can be started.
By the way, the cross-unit adopt() didn't work in Perl until I wrote this example. There was a check against it (the C++ API never bothered with this check). But the adoption between the units has turned out to be quite convenient, so I've removed that check.
And that's it. Long and winding but finally completed. It's kind of about only three lines of code, but I think the rest of it also shows the useful techniques of the work with threads.
TQL server with multithreading
The next big example I've been talking about is finally ready. It's the adaptation of the TQL to work with the multithreaded server framework. The big reason for this example is the export of the table types through a nexus and creation of tables from them. And we'll get to that, but first let's look at the new abilities of the TQL.
TQL is still not of a production quality, in either single- or multi-threaded variety, and contains a large number of simplifying assumptions in its code. As the single-threaded version works symbiotically with the SimpleServer, the multithreaded version works with the ThreadedServer.
One thread created by the programmer contains the "core logic" of the model. It doesn't technically have to be all in a single thread: the data can be forwarded to the other threads and then the results forwarded back from them. But a single core logic thread is a convenient simplification. This thread has some input labels, to receive data from the outside, and some tables with the computed results that can be read by TQL. Of course, it's entirely realistic to have also just the output labels without tables, sending a stream or computed rowops, but again for simplicity I've left this out for now.
This core logic thread creates a TQL instance, which listens on a socket, accepts the connections, forwards the input data to the core logic, performs queries on the tables from the core logic and sends the results back to the client. To this end, the TQL instance creates a few nexuses in the core logic thread and uses them to communicate between all the fragments. The input labels and tables in the core thread also get properly connected to these nexuses. The following figure shows the thread architecture, I'll use it for the reference throughout the discussion:
The core logic thread then goes into its main loop and performs as its name says the core logic computations.
Here is a very simple example of a TQL application:
sub appCoreT # (@opts)
{
my $opts = {};
&Triceps::Opt::parse("appCoreT", $opts, {@Triceps::Triead::opts,
socketName => [ undef, \&Triceps::Opt::ck_mandatory ],
}, @_);
undef @_; # avoids a leak in threads module
my $owner = $opts->{owner};
my $app = $owner->app();
my $unit = $owner->unit();
# build the core logic
my $rtTrade = Triceps::RowType->new(
id => "int32", # trade unique id
symbol => "string", # symbol traded
price => "float64",
size => "float64", # number of shares traded
) or confess "$!";
my $ttWindow = Triceps::TableType->new($rtTrade)
->addSubIndex("byId",
Triceps::SimpleOrderedIndex->new(id => "ASC")
)
or confess "$!";
$ttWindow->initialize() or confess "$!";
# Represents the static information about a company.
my $rtSymbol = Triceps::RowType->new(
symbol => "string", # symbol name
name => "string", # the official company name
eps => "float64", # last quarter earnings per share
) or confess "$!";
my $ttSymbol = Triceps::TableType->new($rtSymbol)
->addSubIndex("bySymbol",
Triceps::SimpleOrderedIndex->new(symbol => "ASC")
)
or confess "$!";
$ttSymbol->initialize() or confess "$!";
my $tWindow = $unit->makeTable($ttWindow, "EM_CALL", "tWindow")
or confess "$!";
my $tSymbol = $unit->makeTable($ttSymbol, "EM_CALL", "tSymbol")
or confess "$!";
# export the endpoints for TQL (it starts the listener)
my $tql = Triceps::X::Tql->new(
name => "tql",
trieadOwner => $owner,
socketName => $opts->{socketName},
tables => [
$tWindow,
$tSymbol,
],
tableNames => [
"window",
"symbol",
],
inputs => [
$tWindow->getInputLabel(),
$tSymbol->getInputLabel(),
],
inputNames => [
"window",
"symbol",
],
);
$owner->readyReady();
$owner->mainLoop();
}
{
my ($port, $thread) = Triceps::X::ThreadedServer::startServer(
app => "appTql",
main => \&appCoreT,
port => 0,
fork => -1, # create a thread, not a process
);
}
This core logic is very simple: all it does is create two tables and then send the input data into them. The server gets started in a background thread (fork => -1) because this code is taken from a test that then goes and runs the expect with the SimpleClient.
The specification of inputs and tables for TQL is somewhat ugly but I kept it as it was historic (it was done this way to keep the parsing of the options simpler). The new options compared to the single-threaded TQL are the "threadOwner", "inputs" and "inputNames". The "threadOwner" is how TQL knows that it must run in the multithreaded mode, and it's used to create the nexuses for communication between the core logic and the rest of TQL. The inputs are needed because the multithreaded TQL parses and forwards the input data, unlike the single-threaded version that relies on the SimpleServer to do that according to the user-defined dispatch table.
The names options don't have to be used: if you name your labels and tables nicely and suitable for the external vieweing, the renaming-for-export can be skipped.
Similar to the single-threaded version, if any of the options "tables" or "inputs" is used, the TQL object gets initialized automatically, otherwise the tables and inputs can be added piecemeal with addTable(), addNamedTable(), addInput(), addNamedInput(), and then the whole thing initialized manually.
Then the clients can establish the connections with the TQL server, send in the data and the queries. To jump in, here is a trace of a simple session that sends some data, then does some table dumps and subscribes, not touching the queries yet. I'll go through it fragment by fragment and explain the meaning. The dumps and subscribes were the warm-up exercises before writing the full queries, but they're useful in their own right, and here they serve as the warm-up exercises for the making of the queries!
> connect c1
c1|ready
The "connect" is not an actual command send but just the indication in the trace that the connection was set up by the client "c1" (it's a trace from the SimpleClient, so it follows the usual conventions). The "ready" response is set when the connection is opened, similar to the chat server shown before.
> c1|subscribe,s1,symbol
c1|subscribe,s1,symbol
This is a subscription request. It means "I'm not interested in the current state of a table but send me all the updates". The response is the mirror of the request, so that the client knows that the request has been processed. "s1" is the unique identifier of the request, so that the client can match together the responses it received to the requests it sent (and keeping the uniqueness is up to the client, the server may refuse the requests with duplicate identifiers). And "symbol" is the name of the table. Once a subscription is in place, there is no way to unsubscribe other than by disconnecting the client (it's doable but adds complications, and I wanted to skip over the nonessential parts). Subscribing multiple times to the same table will send a confirmation every time but the repeated confirmations will have no effect: only one copy of the data will be sent anyway.
> c1|d,symbol,OP_INSERT,ABC,ABC Corp,1.0
c1|d,symbol,OP_INSERT,ABC,ABC Corp,1
This sends the data into the model. And since it propagates through the subscription, the data gets sent back too. The "symbol" here means two different things: on the input side it's the name of the label where the data is sent, on the output side it's the name of the table that has been subscribed to.
The data lines start with the command "d" (since the data is sent much more frequently than the commands, I've picked a short one-letter "command name" for it), then the label/table name, opcode and the row fields in CSV format.
> c1|confirm,cf1
c1|confirm,cf1,,,
The "confirm" command provides a way for the client to check that the data it send had propagated through the model. And it doesn't have to subscribe back to the data and read them. Send some data lines, then send the "confirm" command and wait for it to come back (again, the unique id allows to keep multiple confirmations in flight if you please). This command doesn't guarantee that all the clients have seen the results from that data. It only guarantees that the core logic had seen the data, and more weakly guarantees that the data has been processed by the core logic, and this particular client had already seen all the results from it.
Why weakly? It has to do with the way it works inside, and it depends on the core logic. If the core logic consists of one thread, the guarantee is quite strong. But if the core logic farms out the work from the main thread to the other threads and then collects the results back, the guarantee breaks.
On the Fig. 1 you can see that unlike the chat server shown before, TQL doesn't have any private nexuses for communication between the reader and writer threads of a client. Instead it relies on the same input and output nexuses, adding a control label to them, to forward the commands from the reader to the writer. The TQL object in the core logic thread creates a short-circuit connection between the control labels in the input and output nexuses, forwarding the commands. And if the core logic all runs in one thread, this creates a natural pipeline: the data comes in, gets processed, comes out, the "confirm" command comes in, comes out after the data. But if the core logic farms out the work to more threads, the confirmation can "jump the line" because its path is a direct short circuit.
> c1|drain,dr1
c1|drain,dr1,,,
The "drain" is an analog of "confirm" but more reliable and slower: the reader thread drains the whole model before sending the command on. This guarantees that all the processing is done, and all the output from it has been sent to all the clients.
> c1|dump,d2,symbol
c1|startdump,d2,symbol
c1|d,symbol,OP_INSERT,ABC,ABC Corp,1
c1|dump,d2,symbol
The "dump" command dumps the current contents of a table. Its result starts with "startdump", and the same id and table name as in the request, then goes the data (all with OP_INSERT), finishing with the completion confirmation echoing the original command. The dump is atomic, the contents of the table doesn't change in the middle of the dump. However if a subscription on this table is active, the data rows from that subscription may come before and after the dump.
I'm not going to describe the error reporting, but it's worth mentioning that if a command contains errors, its "confirmation" will be an error line with the same identifier.
> c1|dumpsub,ds3,symbol
c1|startdump,ds3,symbol
c1|d,symbol,OP_INSERT,ABC,ABC Corp,1
c1|dumpsub,ds3,symbol
The "dumpsub" command is a combination of a dump and subscribe: get the initial state and then get all the updates. The confirmation of "dumpsub" marks the boundary between the original dump and the following updates.
> c1|d,symbol,OP_INSERT,DEF,Defense Corp,2.0
c1|d,symbol,OP_INSERT,DEF,Defense Corp,2
Send some more data, and it comes back only once, even though the subscription was done twice: once in "subscribe" and once in "dumpsub". The repeated subscription requests simply get consumed into one subscription.
> c1|d,window,OP_INSERT,1,ABC,101,10
This sends a row to the other table but nothing comes back because there is no subscription to that table.
> c1|dumpsub,ds4,window
c1|startdump,ds4,window
c1|d,window,OP_INSERT,1,ABC,101,10
c1|dumpsub,ds4,window
> c1|d,window,OP_INSERT,2,ABC,102,12
c1|d,window,OP_INSERT,2,ABC,102,12
This demonstrates the pure dump-and-subscribe without any interventions.
> c1|shutdown
c1|shutdown,,,,
c1|__EOF__
And the shutdown command works the same as in the chat server, draning and then shutting down the whole server.
Now on to the queries.
> connect c1
c1|ready
> c1|d,symbol,OP_INSERT,ABC,ABC Corp,1.0
Starts a client connection and sends some data.
> c1|querysub,q1,query1,{read table symbol}{print tokenized 0}
c1|d,query1,OP_INSERT,ABC,ABC Corp,1
c1|querysub,q1,query1
The "querysub" command does the "query-and-subscribe": reads the initial state of the table, processed through the query, and then subscribes to any future updates. The single-threaded variety of TQL doesn't do this, it does just the one-time queries. The multithreaded TQL could also do the one-time queries, and also just the subscribes without the initial state, but I've been cutting corners for this example and the only thing that's actually available is the combination of two, the "querysub".
"q1" is similar to the other command, the command identifier. The next field "query1" is the name for the query, it's the name that will be shown for the data lines coming out of the query. And then goes the query in the brace-quoted format, same as the single-threaded TQL (and there is no further splitting by commas, so the commas can be used freely in the query).
The identified and the name for the query sound kind of redundant. But the client may generate them in different ways and need both. The name has the more symbolic character. The identifier can be generated as a sequence of numbers, so that the client can keep track of its progress more easily. And the error reports include the identifier but not the query name in them.
For the query, there is no special line coming out before the initial dump. Supposedly, there would not be more than one query in flight with the same name, so this could be easily told apart based on the name in the data lines. There is also an underlying consideration that when the query involves a join, in the future the initial dump might be happening in multiple chunks, requiring to either surround every chunk with the start-end lines or just let them go without the extra notifications, as they are now.
And the initial dump ends as usual with getting the echo of the command (without the query part) back.
This particular query is very simple and equivalent to a "dumpsub".
> c1|d,symbol,OP_INSERT,DEF,Defense Corp,2.0
c1|d,query1,OP_INSERT,DEF,Defense Corp,2
Send more data and it will come out of the query.
> c1|querysub,q2,query2,{read table symbol}{where istrue {$%symbol =~ /^A/}}{project fields {symbol eps}}
c1|t,query2,query2 OP_INSERT symbol="ABC" eps="1"
c1|querysub,q2,query2
This query is more complicated, doing a selection (the "where" query command) and projection. It also prints the results in the tokenized format (the "print" command gets added automatically if it wasn't used explicitly, and the default options for it enable the tokenized format).
The tokenized lines come out with the command "t", query name and then the contents of the row. The query name happens to be sent twice, and I'm not sure yet if it's a feature or a bug.
> c1|d,symbol,OP_INSERT,AAA,Absolute Auto Analytics Inc,3.0
c1|d,query1,OP_INSERT,AAA,Absolute Auto Analytics Inc,3
c1|t,query2,query2 OP_INSERT symbol="AAA" eps="3"
> c1|d,symbol,OP_DELETE,DEF,Defense Corp,2.0
c1|d,query1,OP_DELETE,DEF,Defense Corp,2
More examples of the data sent, getting processed by both queries. In the second case the "where" filters out the row from query2, so only query1 produces the result.
> c1|shutdown
c1|shutdown,,,,
c1|__EOF__
And the shutdown as usual.
Now the "piece de resistance": queries with joins.
> connect c1
c1|ready
> c1|d,symbol,OP_INSERT,ABC,ABC Corp,2.0
> c1|d,symbol,OP_INSERT,DEF,Defense Corp,2.0
> c1|d,symbol,OP_INSERT,AAA,Absolute Auto Analytics Inc,3.0
> c1|d,window,OP_INSERT,1,AAA,12,100
Connect and send some starting data.
> c1|querysub,q1,query1,{read table window}{join table symbol byLeft {symbol} type left}
c1|t,query1,query1 OP_INSERT id="1" symbol="AAA" price="12" size="100" name="Absolute Auto Analytics Inc" eps="3"
c1|querysub,q1,query1
A left join of the tables "window" and "symbol", by the field "symbol" as join condition.
Note that unlike the previous single-threaded TQL examples, the index type path for the table "symbol" is not explicitly specified. It's the result of the new method TableType::findIndexPathForKeys() described before, now the index gets found automatically. And the single-threaded TQL now has this feature too. If you really want, you can still specify the index path but usually there is no need to.
The TQL joins, even in the multithreaded mode, are still implemented internally as LookupJoin, driven only by the main flow of the query. So the changes to the joined dimension tables will not update the query results, and will be visible only when a change on the main flow picks them up, potentially creating inconsistencies in the output. This is wrong, but fixing it presents complexities that I've left alone until some later time.
> c1|d,window,OP_INSERT,2,ABC,13,100
c1|t,query1,query1 OP_INSERT id="2" symbol="ABC" price="13" size="100" name="ABC Corp" eps="2"
> c1|d,window,OP_INSERT,3,AAA,11,200
c1|t,query1,query1 OP_INSERT id="3" symbol="AAA" price="11" size="200" name="Absolute Auto Analytics Inc" eps="3"
Sending data updates the results of the query.
> c1|d,symbol,OP_DELETE,AAA,Absolute Auto Analytics Inc,3.0
> c1|d,symbol,OP_INSERT,AAA,Alcoholic Abstract Aliens,3.0
As described above, the modifications of the dimension table are mot visible in the query directly.
> c1|d,window,OP_DELETE,1
c1|t,query1,query1 OP_DELETE id="1" symbol="AAA" price="12" size="100" name="Alcoholic Abstract Aliens" eps="3"
But an update on the main flow brings them up (an in this case inconsistently, the row getting deleted is not exactly the same as the row inserted before).
> c1|querysub,q2,query2,{read table window}{join table symbol byLeft {symbol} type left}{join table symbol byLeft {eps} type left rightFields {symbol/symbol2}}
c1|t,query2,query2 OP_INSERT id="2" symbol="ABC" price="13" size="100" name="ABC Corp" eps="2" symbol2="ABC"
c1|t,query2,query2 OP_INSERT id="2" symbol="ABC" price="13" size="100" name="ABC Corp" eps="2" symbol2="DEF"
c1|t,query2,query2 OP_INSERT id="3" symbol="AAA" price="11" size="200" name="Alcoholic Abstract Aliens" eps="3" symbol2="AAA"
c1|querysub,q2,query2
This is a more complicated query, involving two joins, with the same dimension table "symbol". The second join by "eps" makes no real-world sense whatsoever but it's interesting from the technical perspective: if you check the table type of this table at the start of the post, you'll find that it has no index on the field "eps". The join adds this index on demand!
The way it works, all the dimension tables are copied into the client's writer thread, created from the table types exported by the core logic throuhg the output nexus. (And if a table is used in the same query twice, it's currently also copied twice). This provides a nice opportunity to amend the table type by adding any necessary secondary index before creating the table, and TQL makes a good use of it.
The details are forthcoming in the next post.
TQL is still not of a production quality, in either single- or multi-threaded variety, and contains a large number of simplifying assumptions in its code. As the single-threaded version works symbiotically with the SimpleServer, the multithreaded version works with the ThreadedServer.
One thread created by the programmer contains the "core logic" of the model. It doesn't technically have to be all in a single thread: the data can be forwarded to the other threads and then the results forwarded back from them. But a single core logic thread is a convenient simplification. This thread has some input labels, to receive data from the outside, and some tables with the computed results that can be read by TQL. Of course, it's entirely realistic to have also just the output labels without tables, sending a stream or computed rowops, but again for simplicity I've left this out for now.
This core logic thread creates a TQL instance, which listens on a socket, accepts the connections, forwards the input data to the core logic, performs queries on the tables from the core logic and sends the results back to the client. To this end, the TQL instance creates a few nexuses in the core logic thread and uses them to communicate between all the fragments. The input labels and tables in the core thread also get properly connected to these nexuses. The following figure shows the thread architecture, I'll use it for the reference throughout the discussion:
 |
| Fig. 1. TQL application. |
The core logic thread then goes into its main loop and performs as its name says the core logic computations.
Here is a very simple example of a TQL application:
sub appCoreT # (@opts)
{
my $opts = {};
&Triceps::Opt::parse("appCoreT", $opts, {@Triceps::Triead::opts,
socketName => [ undef, \&Triceps::Opt::ck_mandatory ],
}, @_);
undef @_; # avoids a leak in threads module
my $owner = $opts->{owner};
my $app = $owner->app();
my $unit = $owner->unit();
# build the core logic
my $rtTrade = Triceps::RowType->new(
id => "int32", # trade unique id
symbol => "string", # symbol traded
price => "float64",
size => "float64", # number of shares traded
) or confess "$!";
my $ttWindow = Triceps::TableType->new($rtTrade)
->addSubIndex("byId",
Triceps::SimpleOrderedIndex->new(id => "ASC")
)
or confess "$!";
$ttWindow->initialize() or confess "$!";
# Represents the static information about a company.
my $rtSymbol = Triceps::RowType->new(
symbol => "string", # symbol name
name => "string", # the official company name
eps => "float64", # last quarter earnings per share
) or confess "$!";
my $ttSymbol = Triceps::TableType->new($rtSymbol)
->addSubIndex("bySymbol",
Triceps::SimpleOrderedIndex->new(symbol => "ASC")
)
or confess "$!";
$ttSymbol->initialize() or confess "$!";
my $tWindow = $unit->makeTable($ttWindow, "EM_CALL", "tWindow")
or confess "$!";
my $tSymbol = $unit->makeTable($ttSymbol, "EM_CALL", "tSymbol")
or confess "$!";
# export the endpoints for TQL (it starts the listener)
my $tql = Triceps::X::Tql->new(
name => "tql",
trieadOwner => $owner,
socketName => $opts->{socketName},
tables => [
$tWindow,
$tSymbol,
],
tableNames => [
"window",
"symbol",
],
inputs => [
$tWindow->getInputLabel(),
$tSymbol->getInputLabel(),
],
inputNames => [
"window",
"symbol",
],
);
$owner->readyReady();
$owner->mainLoop();
}
{
my ($port, $thread) = Triceps::X::ThreadedServer::startServer(
app => "appTql",
main => \&appCoreT,
port => 0,
fork => -1, # create a thread, not a process
);
}
This core logic is very simple: all it does is create two tables and then send the input data into them. The server gets started in a background thread (fork => -1) because this code is taken from a test that then goes and runs the expect with the SimpleClient.
The specification of inputs and tables for TQL is somewhat ugly but I kept it as it was historic (it was done this way to keep the parsing of the options simpler). The new options compared to the single-threaded TQL are the "threadOwner", "inputs" and "inputNames". The "threadOwner" is how TQL knows that it must run in the multithreaded mode, and it's used to create the nexuses for communication between the core logic and the rest of TQL. The inputs are needed because the multithreaded TQL parses and forwards the input data, unlike the single-threaded version that relies on the SimpleServer to do that according to the user-defined dispatch table.
The names options don't have to be used: if you name your labels and tables nicely and suitable for the external vieweing, the renaming-for-export can be skipped.
Similar to the single-threaded version, if any of the options "tables" or "inputs" is used, the TQL object gets initialized automatically, otherwise the tables and inputs can be added piecemeal with addTable(), addNamedTable(), addInput(), addNamedInput(), and then the whole thing initialized manually.
Then the clients can establish the connections with the TQL server, send in the data and the queries. To jump in, here is a trace of a simple session that sends some data, then does some table dumps and subscribes, not touching the queries yet. I'll go through it fragment by fragment and explain the meaning. The dumps and subscribes were the warm-up exercises before writing the full queries, but they're useful in their own right, and here they serve as the warm-up exercises for the making of the queries!
> connect c1
c1|ready
The "connect" is not an actual command send but just the indication in the trace that the connection was set up by the client "c1" (it's a trace from the SimpleClient, so it follows the usual conventions). The "ready" response is set when the connection is opened, similar to the chat server shown before.
> c1|subscribe,s1,symbol
c1|subscribe,s1,symbol
This is a subscription request. It means "I'm not interested in the current state of a table but send me all the updates". The response is the mirror of the request, so that the client knows that the request has been processed. "s1" is the unique identifier of the request, so that the client can match together the responses it received to the requests it sent (and keeping the uniqueness is up to the client, the server may refuse the requests with duplicate identifiers). And "symbol" is the name of the table. Once a subscription is in place, there is no way to unsubscribe other than by disconnecting the client (it's doable but adds complications, and I wanted to skip over the nonessential parts). Subscribing multiple times to the same table will send a confirmation every time but the repeated confirmations will have no effect: only one copy of the data will be sent anyway.
> c1|d,symbol,OP_INSERT,ABC,ABC Corp,1.0
c1|d,symbol,OP_INSERT,ABC,ABC Corp,1
This sends the data into the model. And since it propagates through the subscription, the data gets sent back too. The "symbol" here means two different things: on the input side it's the name of the label where the data is sent, on the output side it's the name of the table that has been subscribed to.
The data lines start with the command "d" (since the data is sent much more frequently than the commands, I've picked a short one-letter "command name" for it), then the label/table name, opcode and the row fields in CSV format.
> c1|confirm,cf1
c1|confirm,cf1,,,
The "confirm" command provides a way for the client to check that the data it send had propagated through the model. And it doesn't have to subscribe back to the data and read them. Send some data lines, then send the "confirm" command and wait for it to come back (again, the unique id allows to keep multiple confirmations in flight if you please). This command doesn't guarantee that all the clients have seen the results from that data. It only guarantees that the core logic had seen the data, and more weakly guarantees that the data has been processed by the core logic, and this particular client had already seen all the results from it.
Why weakly? It has to do with the way it works inside, and it depends on the core logic. If the core logic consists of one thread, the guarantee is quite strong. But if the core logic farms out the work from the main thread to the other threads and then collects the results back, the guarantee breaks.
On the Fig. 1 you can see that unlike the chat server shown before, TQL doesn't have any private nexuses for communication between the reader and writer threads of a client. Instead it relies on the same input and output nexuses, adding a control label to them, to forward the commands from the reader to the writer. The TQL object in the core logic thread creates a short-circuit connection between the control labels in the input and output nexuses, forwarding the commands. And if the core logic all runs in one thread, this creates a natural pipeline: the data comes in, gets processed, comes out, the "confirm" command comes in, comes out after the data. But if the core logic farms out the work to more threads, the confirmation can "jump the line" because its path is a direct short circuit.
> c1|drain,dr1
c1|drain,dr1,,,
The "drain" is an analog of "confirm" but more reliable and slower: the reader thread drains the whole model before sending the command on. This guarantees that all the processing is done, and all the output from it has been sent to all the clients.
> c1|dump,d2,symbol
c1|startdump,d2,symbol
c1|d,symbol,OP_INSERT,ABC,ABC Corp,1
c1|dump,d2,symbol
The "dump" command dumps the current contents of a table. Its result starts with "startdump", and the same id and table name as in the request, then goes the data (all with OP_INSERT), finishing with the completion confirmation echoing the original command. The dump is atomic, the contents of the table doesn't change in the middle of the dump. However if a subscription on this table is active, the data rows from that subscription may come before and after the dump.
I'm not going to describe the error reporting, but it's worth mentioning that if a command contains errors, its "confirmation" will be an error line with the same identifier.
> c1|dumpsub,ds3,symbol
c1|startdump,ds3,symbol
c1|d,symbol,OP_INSERT,ABC,ABC Corp,1
c1|dumpsub,ds3,symbol
The "dumpsub" command is a combination of a dump and subscribe: get the initial state and then get all the updates. The confirmation of "dumpsub" marks the boundary between the original dump and the following updates.
> c1|d,symbol,OP_INSERT,DEF,Defense Corp,2.0
c1|d,symbol,OP_INSERT,DEF,Defense Corp,2
Send some more data, and it comes back only once, even though the subscription was done twice: once in "subscribe" and once in "dumpsub". The repeated subscription requests simply get consumed into one subscription.
> c1|d,window,OP_INSERT,1,ABC,101,10
This sends a row to the other table but nothing comes back because there is no subscription to that table.
> c1|dumpsub,ds4,window
c1|startdump,ds4,window
c1|d,window,OP_INSERT,1,ABC,101,10
c1|dumpsub,ds4,window
> c1|d,window,OP_INSERT,2,ABC,102,12
c1|d,window,OP_INSERT,2,ABC,102,12
This demonstrates the pure dump-and-subscribe without any interventions.
> c1|shutdown
c1|shutdown,,,,
c1|__EOF__
And the shutdown command works the same as in the chat server, draning and then shutting down the whole server.
Now on to the queries.
> connect c1
c1|ready
> c1|d,symbol,OP_INSERT,ABC,ABC Corp,1.0
Starts a client connection and sends some data.
> c1|querysub,q1,query1,{read table symbol}{print tokenized 0}
c1|d,query1,OP_INSERT,ABC,ABC Corp,1
c1|querysub,q1,query1
The "querysub" command does the "query-and-subscribe": reads the initial state of the table, processed through the query, and then subscribes to any future updates. The single-threaded variety of TQL doesn't do this, it does just the one-time queries. The multithreaded TQL could also do the one-time queries, and also just the subscribes without the initial state, but I've been cutting corners for this example and the only thing that's actually available is the combination of two, the "querysub".
"q1" is similar to the other command, the command identifier. The next field "query1" is the name for the query, it's the name that will be shown for the data lines coming out of the query. And then goes the query in the brace-quoted format, same as the single-threaded TQL (and there is no further splitting by commas, so the commas can be used freely in the query).
The identified and the name for the query sound kind of redundant. But the client may generate them in different ways and need both. The name has the more symbolic character. The identifier can be generated as a sequence of numbers, so that the client can keep track of its progress more easily. And the error reports include the identifier but not the query name in them.
For the query, there is no special line coming out before the initial dump. Supposedly, there would not be more than one query in flight with the same name, so this could be easily told apart based on the name in the data lines. There is also an underlying consideration that when the query involves a join, in the future the initial dump might be happening in multiple chunks, requiring to either surround every chunk with the start-end lines or just let them go without the extra notifications, as they are now.
And the initial dump ends as usual with getting the echo of the command (without the query part) back.
This particular query is very simple and equivalent to a "dumpsub".
> c1|d,symbol,OP_INSERT,DEF,Defense Corp,2.0
c1|d,query1,OP_INSERT,DEF,Defense Corp,2
Send more data and it will come out of the query.
> c1|querysub,q2,query2,{read table symbol}{where istrue {$%symbol =~ /^A/}}{project fields {symbol eps}}
c1|t,query2,query2 OP_INSERT symbol="ABC" eps="1"
c1|querysub,q2,query2
This query is more complicated, doing a selection (the "where" query command) and projection. It also prints the results in the tokenized format (the "print" command gets added automatically if it wasn't used explicitly, and the default options for it enable the tokenized format).
The tokenized lines come out with the command "t", query name and then the contents of the row. The query name happens to be sent twice, and I'm not sure yet if it's a feature or a bug.
> c1|d,symbol,OP_INSERT,AAA,Absolute Auto Analytics Inc,3.0
c1|d,query1,OP_INSERT,AAA,Absolute Auto Analytics Inc,3
c1|t,query2,query2 OP_INSERT symbol="AAA" eps="3"
> c1|d,symbol,OP_DELETE,DEF,Defense Corp,2.0
c1|d,query1,OP_DELETE,DEF,Defense Corp,2
More examples of the data sent, getting processed by both queries. In the second case the "where" filters out the row from query2, so only query1 produces the result.
> c1|shutdown
c1|shutdown,,,,
c1|__EOF__
And the shutdown as usual.
Now the "piece de resistance": queries with joins.
> connect c1
c1|ready
> c1|d,symbol,OP_INSERT,ABC,ABC Corp,2.0
> c1|d,symbol,OP_INSERT,DEF,Defense Corp,2.0
> c1|d,symbol,OP_INSERT,AAA,Absolute Auto Analytics Inc,3.0
> c1|d,window,OP_INSERT,1,AAA,12,100
Connect and send some starting data.
> c1|querysub,q1,query1,{read table window}{join table symbol byLeft {symbol} type left}
c1|t,query1,query1 OP_INSERT id="1" symbol="AAA" price="12" size="100" name="Absolute Auto Analytics Inc" eps="3"
c1|querysub,q1,query1
A left join of the tables "window" and "symbol", by the field "symbol" as join condition.
Note that unlike the previous single-threaded TQL examples, the index type path for the table "symbol" is not explicitly specified. It's the result of the new method TableType::findIndexPathForKeys() described before, now the index gets found automatically. And the single-threaded TQL now has this feature too. If you really want, you can still specify the index path but usually there is no need to.
The TQL joins, even in the multithreaded mode, are still implemented internally as LookupJoin, driven only by the main flow of the query. So the changes to the joined dimension tables will not update the query results, and will be visible only when a change on the main flow picks them up, potentially creating inconsistencies in the output. This is wrong, but fixing it presents complexities that I've left alone until some later time.
> c1|d,window,OP_INSERT,2,ABC,13,100
c1|t,query1,query1 OP_INSERT id="2" symbol="ABC" price="13" size="100" name="ABC Corp" eps="2"
> c1|d,window,OP_INSERT,3,AAA,11,200
c1|t,query1,query1 OP_INSERT id="3" symbol="AAA" price="11" size="200" name="Absolute Auto Analytics Inc" eps="3"
Sending data updates the results of the query.
> c1|d,symbol,OP_DELETE,AAA,Absolute Auto Analytics Inc,3.0
> c1|d,symbol,OP_INSERT,AAA,Alcoholic Abstract Aliens,3.0
As described above, the modifications of the dimension table are mot visible in the query directly.
> c1|d,window,OP_DELETE,1
c1|t,query1,query1 OP_DELETE id="1" symbol="AAA" price="12" size="100" name="Alcoholic Abstract Aliens" eps="3"
But an update on the main flow brings them up (an in this case inconsistently, the row getting deleted is not exactly the same as the row inserted before).
> c1|querysub,q2,query2,{read table window}{join table symbol byLeft {symbol} type left}{join table symbol byLeft {eps} type left rightFields {symbol/symbol2}}
c1|t,query2,query2 OP_INSERT id="2" symbol="ABC" price="13" size="100" name="ABC Corp" eps="2" symbol2="ABC"
c1|t,query2,query2 OP_INSERT id="2" symbol="ABC" price="13" size="100" name="ABC Corp" eps="2" symbol2="DEF"
c1|t,query2,query2 OP_INSERT id="3" symbol="AAA" price="11" size="200" name="Alcoholic Abstract Aliens" eps="3" symbol2="AAA"
c1|querysub,q2,query2
This is a more complicated query, involving two joins, with the same dimension table "symbol". The second join by "eps" makes no real-world sense whatsoever but it's interesting from the technical perspective: if you check the table type of this table at the start of the post, you'll find that it has no index on the field "eps". The join adds this index on demand!
The way it works, all the dimension tables are copied into the client's writer thread, created from the table types exported by the core logic throuhg the output nexus. (And if a table is used in the same query twice, it's currently also copied twice). This provides a nice opportunity to amend the table type by adding any necessary secondary index before creating the table, and TQL makes a good use of it.
The details are forthcoming in the next post.
Friday, May 24, 2013
how to find an index
When working on the next example (still in progress but now getting closer to completion), I've solved one more nagging little problem: finding an index type in the table type by key fields. It makes the joins more convenient: if you know, by which fields you want to join, it's nice to find the correct index automatically. My previous solution to this problem has been to go in the opposite direction: specify the indexes for the join, and find the list of fields automatically from there. But that created a problem with deciding, which field on the left to pair with which on the right, and so either the indexes had to be carefully controlled to have the key fields in the same order, or a separate option still had to be used to specify the pairing of the fields.
Now this problem is solved with the method:
@idxPath = $tableType-> findIndexPathForKeys(@keyFields);
It returns the array that represents the path to an index type that matches these key fields (the index type and all the types in the path still have to be of the Hashed variety). If the correct index can not be found, an empty array is returned. If you specify the fields that aren't present in the row type in the first place, this is simply treated the same as being unable to find an index for these fields.
If more that one index would match, the first one found in the direct order of the index tree walk is returned.
This allowed to change the LookupJoin to make the option "rightIdxPath" optional: if undefined, the index that matches the key fields specified in the option "by" or "byLeft" will be found automatically. Of course, if such an index doesn't exist, it's still an error, the join could not create it for you yet.
The same way, in the JoinTwo now the options "leftIdxPath" and "rightIdxPath" have become optional if either "by" or "byLeft" is specified. The old way of specifying "leftIdxPath" and "rightIdxPath" without any of the "by" varieties still works too.
And this propagated into the TQL command "join": no more need for the option "rightIdxPath". You can still use it if you want but there is rarely any point to it, only if there are multiple matching indexes and you strongly prefer one of them.
One more indirect fall-out from these changes is the new option to LookupJoin:
fieldsDropRightKey => 0/1
The default is 0, and if you set it to 1, the logic will automatically exclude from the result row type the key fields from the right side. This is convenient because these fields contain the values that are duplicates of the key fields from the left side anyway. It's what the TQL join was doing all along, and the implementation of this logic becomes simpler when it gets moved directly into LookupJoin.
Now this problem is solved with the method:
@idxPath = $tableType-> findIndexPathForKeys(@keyFields);
It returns the array that represents the path to an index type that matches these key fields (the index type and all the types in the path still have to be of the Hashed variety). If the correct index can not be found, an empty array is returned. If you specify the fields that aren't present in the row type in the first place, this is simply treated the same as being unable to find an index for these fields.
If more that one index would match, the first one found in the direct order of the index tree walk is returned.
This allowed to change the LookupJoin to make the option "rightIdxPath" optional: if undefined, the index that matches the key fields specified in the option "by" or "byLeft" will be found automatically. Of course, if such an index doesn't exist, it's still an error, the join could not create it for you yet.
The same way, in the JoinTwo now the options "leftIdxPath" and "rightIdxPath" have become optional if either "by" or "byLeft" is specified. The old way of specifying "leftIdxPath" and "rightIdxPath" without any of the "by" varieties still works too.
And this propagated into the TQL command "join": no more need for the option "rightIdxPath". You can still use it if you want but there is rarely any point to it, only if there are multiple matching indexes and you strongly prefer one of them.
One more indirect fall-out from these changes is the new option to LookupJoin:
fieldsDropRightKey => 0/1
The default is 0, and if you set it to 1, the logic will automatically exclude from the result row type the key fields from the right side. This is convenient because these fields contain the values that are duplicates of the key fields from the left side anyway. It's what the TQL join was doing all along, and the implementation of this logic becomes simpler when it gets moved directly into LookupJoin.
Monday, November 26, 2012
Streaming functions and unit boundaries (and TQL guts)
Now let's take a look at the insides of the Tql module. I'll be skipping over the code that is less interesting, you can find the full version in the source code of perl/Triceps/lib/Triceps/X/Tql.pm as always. The constructor is one of these things to be skipped. The initialization part is more interesting:
sub initialize # ($self)
{
my $myname = "Triceps::X::Tql::initialize";
my $self = shift;
return if ($self->{initialized});
my %dispatch;
my @labels;
for (my $i = 0; $i <= $#{$self->{tables}}; $i++) {
my $name = $self->{tableNames}[$i];
my $table = $self->{tables}[$i];
confess "$myname: found a duplicate table name '$name', all names are: "
. join(", ", @{$self->{tableNames}})
if (exists $dispatch{$name});
$dispatch{$name} = $table;
push @labels, $name, $table->getDumpLabel();
}
$self->{dispatch} = \%dispatch;
$self->{fret} = Triceps::FnReturn->new(
name => $self->{name} . ".fret",
labels => \@labels,
);
$self->{initialized} = 1;
}
It creates a dispatch table of name-to-table and also an FnReturn that contains the dump labels of all the tables.
Each query will be created as its own unit. It will run, and then get cleared and disposed of, very convenient. By the way, that is the answer to the question of why would someone want to use multiple units in the same thread: for modular disposal.
But the labels in the main unit and the query unit can't be directly connected. A direct connection would create the stable references, and the disposal won't work. That's where the streaming function interface comes to the rescue: it provides a temporary connection. Build the query unit, build a binding for it, push the binding onto the FnReturn of the main unit, run the query, pop the binding, dispose of the query unit.
And the special capacity (or if you will, superpower) of the streaming functions that allows all that is that the FnReturn and FnBinding don't have to be of the same unit. They may be of the different units and will still work together fine.
The query() method then handles the creation of the unit and stuff:
sub query # ($self, $argline)
{
my $myname = "Triceps::X::Tql::query";
my $self = shift;
my $argline = shift;
confess "$myname: may be used only on an initialized object"
unless ($self->{initialized});
$argline =~ s/^([^,]*)(,|$)//; # skip the name of the label
my $q = $1; # the name of the query itself
#&Triceps::X::SimpleServer::outCurBuf("+DEBUGquery: $argline\n");
my @cmds = split_braced($argline);
if ($argline ne '') {
# Presumably, the argument line should contain no line feeds, so it should be safe to send back.
&Triceps::X::SimpleServer::outCurBuf("+ERROR,OP_INSERT,$q: mismatched braces in the trailing $argline\n");
return
}
# The context for the commands to build up an execution of a query.
# Unlike $self, the context is created afresh for every query.
my $ctx = {};
# The query will be built in a separate unit
$ctx->{tables} = $self->{dispatch};
$ctx->{fretDumps} = $self->{fret};
$ctx->{u} = Triceps::Unit->new("${q}.unit");
$ctx->{prev} = undef; # will contain the output of the previous command in the pipeline
$ctx->{actions} = []; # code that will run the pipeline
$ctx->{id} = 0; # a unique id for auto-generated objects
# It's important to place the clearing trigger outside eval {}. Otherwise the
# clearing will erase any errors in $@ returned from eval.
my $cleaner = $ctx->{u}->makeClearingTrigger();
if (! eval {
foreach my $cmd (@cmds) {
#&Triceps::X::SimpleServer::outCurBuf("+DEBUGcmd, $cmd\n");
my @args = split_braced($cmd);
my $argv0 = bunquote(shift @args);
# The rest of @args do not get unquoted here!
die "No such TQL command '$argv0'\n" unless exists $tqlDispatch{$argv0};
$ctx->{id}++;
&{$tqlDispatch{$argv0}}($ctx, @args);
# Each command must set its result label (even if an undef) into
# $ctx->{next}.
die "Internal error in the command $argv0: missing result definition\n"
unless (exists $ctx->{next});
$ctx->{prev} = $ctx->{next};
delete $ctx->{next};
}
if (defined $ctx->{prev}) {
# implicitly print the result of the pipeline, no options
&{$tqlDispatch{"print"}}($ctx);
}
# Now run the pipeline
foreach my $code (@{$ctx->{actions}}) {
&$code;
}
# Now run the pipeline
1; # means that everything went OK
}) {
# XXX this won't work well with the multi-line errors
&Triceps::X::SimpleServer::outCurBuf("+ERROR,OP_INSERT,$q: error: $@\n");
return
}
}
Each TQL command is defined as its own method, all of them collected in the %tqlDispatch. query() splits the pipeline and then lets each command build its part of the query, connecting them through $ctx. A command may also register an action to be run later. After everything is built, the actions run and produce the result.
The functions split_braced() and bunquote() are imported from the package Triceps::X::Braced that handles the parsing of the braced nested lists.
Another interesting part is the error reporting, done as a special label "+ERROR". It's actually one of the sticky points of why the code is not of production quality: because the errors may be multi-line, and the SimpleServer protocol really expects everything to be single-line. Properly, some quoting would have to be done.
Moving on, here is the "read" command handler:
sub _tqlRead # ($ctx, @args)
{
my $ctx = shift;
die "The read command may not be used in the middle of a pipeline.\n"
if (defined($ctx->{prev}));
my $opts = {};
&Triceps::Opt::parse("read", $opts, {
table => [ undef, \&Triceps::Opt::ck_mandatory ],
}, @_);
my $fret = $ctx->{fretDumps};
my $tabname = bunquote($opts->{table});
die ("Read found no such table '$tabname'\n")
unless (exists $ctx->{tables}{$tabname});
my $unit = $ctx->{u};
my $table = $ctx->{tables}{$tabname};
my $lab = $unit->makeDummyLabel($table->getRowType(), "lb" . $ctx->{id} . "read");
$ctx->{next} = $lab;
my $code = sub {
Triceps::FnBinding::call(
name => "bind" . $ctx->{id} . "read",
unit => $unit,
on => $fret,
labels => [
$tabname => $lab,
],
code => sub {
$table->dumpAll();
},
);
};
push @{$ctx->{actions}}, $code;
}
It's the only command that registers an action, which sends data into the query unit. The rest of commands just add more handlers to the pipeline in the unit, and get the data that flows from "read". The action sets up a binding and calls the table dump, to send the data into that binding.
The reading of the tables could have also been done without the bindings, and without the need to bind the units at all: just iterate through the table procedurally in the action. But this whole example has been built largely to showcase that the bindings can be used in this way, so naturally it uses bindings.
The bindings come more useful when the query logic has to react to the normal logic of the main unit, such as in the subscriptions: set up the query, read its initial state, and then keep reading as the state gets updated. But guess what, the subscriptions can't be done with the FnReturns as shown because the FnReturn only sends its data to the last binding pushed onto it. This means, if multiple subscriptions get set up, only the last one will be getting the data. There will be a separate mechanism for that.
sub initialize # ($self)
{
my $myname = "Triceps::X::Tql::initialize";
my $self = shift;
return if ($self->{initialized});
my %dispatch;
my @labels;
for (my $i = 0; $i <= $#{$self->{tables}}; $i++) {
my $name = $self->{tableNames}[$i];
my $table = $self->{tables}[$i];
confess "$myname: found a duplicate table name '$name', all names are: "
. join(", ", @{$self->{tableNames}})
if (exists $dispatch{$name});
$dispatch{$name} = $table;
push @labels, $name, $table->getDumpLabel();
}
$self->{dispatch} = \%dispatch;
$self->{fret} = Triceps::FnReturn->new(
name => $self->{name} . ".fret",
labels => \@labels,
);
$self->{initialized} = 1;
}
It creates a dispatch table of name-to-table and also an FnReturn that contains the dump labels of all the tables.
Each query will be created as its own unit. It will run, and then get cleared and disposed of, very convenient. By the way, that is the answer to the question of why would someone want to use multiple units in the same thread: for modular disposal.
But the labels in the main unit and the query unit can't be directly connected. A direct connection would create the stable references, and the disposal won't work. That's where the streaming function interface comes to the rescue: it provides a temporary connection. Build the query unit, build a binding for it, push the binding onto the FnReturn of the main unit, run the query, pop the binding, dispose of the query unit.
And the special capacity (or if you will, superpower) of the streaming functions that allows all that is that the FnReturn and FnBinding don't have to be of the same unit. They may be of the different units and will still work together fine.
The query() method then handles the creation of the unit and stuff:
sub query # ($self, $argline)
{
my $myname = "Triceps::X::Tql::query";
my $self = shift;
my $argline = shift;
confess "$myname: may be used only on an initialized object"
unless ($self->{initialized});
$argline =~ s/^([^,]*)(,|$)//; # skip the name of the label
my $q = $1; # the name of the query itself
#&Triceps::X::SimpleServer::outCurBuf("+DEBUGquery: $argline\n");
my @cmds = split_braced($argline);
if ($argline ne '') {
# Presumably, the argument line should contain no line feeds, so it should be safe to send back.
&Triceps::X::SimpleServer::outCurBuf("+ERROR,OP_INSERT,$q: mismatched braces in the trailing $argline\n");
return
}
# The context for the commands to build up an execution of a query.
# Unlike $self, the context is created afresh for every query.
my $ctx = {};
# The query will be built in a separate unit
$ctx->{tables} = $self->{dispatch};
$ctx->{fretDumps} = $self->{fret};
$ctx->{u} = Triceps::Unit->new("${q}.unit");
$ctx->{prev} = undef; # will contain the output of the previous command in the pipeline
$ctx->{actions} = []; # code that will run the pipeline
$ctx->{id} = 0; # a unique id for auto-generated objects
# It's important to place the clearing trigger outside eval {}. Otherwise the
# clearing will erase any errors in $@ returned from eval.
my $cleaner = $ctx->{u}->makeClearingTrigger();
if (! eval {
foreach my $cmd (@cmds) {
#&Triceps::X::SimpleServer::outCurBuf("+DEBUGcmd, $cmd\n");
my @args = split_braced($cmd);
my $argv0 = bunquote(shift @args);
# The rest of @args do not get unquoted here!
die "No such TQL command '$argv0'\n" unless exists $tqlDispatch{$argv0};
$ctx->{id}++;
&{$tqlDispatch{$argv0}}($ctx, @args);
# Each command must set its result label (even if an undef) into
# $ctx->{next}.
die "Internal error in the command $argv0: missing result definition\n"
unless (exists $ctx->{next});
$ctx->{prev} = $ctx->{next};
delete $ctx->{next};
}
if (defined $ctx->{prev}) {
# implicitly print the result of the pipeline, no options
&{$tqlDispatch{"print"}}($ctx);
}
# Now run the pipeline
foreach my $code (@{$ctx->{actions}}) {
&$code;
}
# Now run the pipeline
1; # means that everything went OK
}) {
# XXX this won't work well with the multi-line errors
&Triceps::X::SimpleServer::outCurBuf("+ERROR,OP_INSERT,$q: error: $@\n");
return
}
}
Each TQL command is defined as its own method, all of them collected in the %tqlDispatch. query() splits the pipeline and then lets each command build its part of the query, connecting them through $ctx. A command may also register an action to be run later. After everything is built, the actions run and produce the result.
The functions split_braced() and bunquote() are imported from the package Triceps::X::Braced that handles the parsing of the braced nested lists.
Another interesting part is the error reporting, done as a special label "+ERROR". It's actually one of the sticky points of why the code is not of production quality: because the errors may be multi-line, and the SimpleServer protocol really expects everything to be single-line. Properly, some quoting would have to be done.
Moving on, here is the "read" command handler:
sub _tqlRead # ($ctx, @args)
{
my $ctx = shift;
die "The read command may not be used in the middle of a pipeline.\n"
if (defined($ctx->{prev}));
my $opts = {};
&Triceps::Opt::parse("read", $opts, {
table => [ undef, \&Triceps::Opt::ck_mandatory ],
}, @_);
my $fret = $ctx->{fretDumps};
my $tabname = bunquote($opts->{table});
die ("Read found no such table '$tabname'\n")
unless (exists $ctx->{tables}{$tabname});
my $unit = $ctx->{u};
my $table = $ctx->{tables}{$tabname};
my $lab = $unit->makeDummyLabel($table->getRowType(), "lb" . $ctx->{id} . "read");
$ctx->{next} = $lab;
my $code = sub {
Triceps::FnBinding::call(
name => "bind" . $ctx->{id} . "read",
unit => $unit,
on => $fret,
labels => [
$tabname => $lab,
],
code => sub {
$table->dumpAll();
},
);
};
push @{$ctx->{actions}}, $code;
}
It's the only command that registers an action, which sends data into the query unit. The rest of commands just add more handlers to the pipeline in the unit, and get the data that flows from "read". The action sets up a binding and calls the table dump, to send the data into that binding.
The reading of the tables could have also been done without the bindings, and without the need to bind the units at all: just iterate through the table procedurally in the action. But this whole example has been built largely to showcase that the bindings can be used in this way, so naturally it uses bindings.
The bindings come more useful when the query logic has to react to the normal logic of the main unit, such as in the subscriptions: set up the query, read its initial state, and then keep reading as the state gets updated. But guess what, the subscriptions can't be done with the FnReturns as shown because the FnReturn only sends its data to the last binding pushed onto it. This means, if multiple subscriptions get set up, only the last one will be getting the data. There will be a separate mechanism for that.
running the TQL query server
The code that produced the query output examples from the previous post looks like this:
# The basic table type to be used for querying.
# Represents the trades reports.
our $rtTrade = Triceps::RowType->new(
id => "int32", # trade unique id
symbol => "string", # symbol traded
price => "float64",
size => "float64", # number of shares traded
) or confess "$!";
our $ttWindow = Triceps::TableType->new($rtTrade)
->addSubIndex("bySymbol",
Triceps::SimpleOrderedIndex->new(symbol => "ASC")
->addSubIndex("last2",
Triceps::IndexType->newFifo(limit => 2)
)
)
or confess "$!";
$ttWindow->initialize() or confess "$!";
# Represents the static information about a company.
our $rtSymbol = Triceps::RowType->new(
symbol => "string", # symbol name
name => "string", # the official company name
eps => "float64", # last quarter earnings per share
) or confess "$!";
our $ttSymbol = Triceps::TableType->new($rtSymbol)
->addSubIndex("bySymbol",
Triceps::IndexType->newHashed(key => [ "symbol" ])
)
or confess "$!";
$ttSymbol->initialize() or confess "$!";
my $uTrades = Triceps::Unit->new("uTrades");
my $tWindow = $uTrades->makeTable($ttWindow, "EM_CALL", "tWindow")
or confess "$!";
my $tSymbol = $uTrades->makeTable($ttSymbol, "EM_CALL", "tSymbol")
or confess "$!";
# The information about tables, for querying.
my $tql = Triceps::X::Tql->new(
name => "tql",
tables => [
$tWindow,
$tSymbol,
],
);
my %dispatch;
$dispatch{$tWindow->getName()} = $tWindow->getInputLabel();
$dispatch{$tSymbol->getName()} = $tSymbol->getInputLabel();
$dispatch{"query"} = sub { $tql->query(@_); };
$dispatch{"exit"} = \&Triceps::X::SimpleServer::exitFunc;
Triceps::X::DumbClient::run(\%dispatch);
It's very much like the example shown before in the section 7.8 "Main loop with a socket", with a few differences. Obviously, Tql has been added, and we'll get to that part just in a moment. But the other differences are centered around the way the server and client code has been restructured.
The Triceps::X::DumbClient is a module for testing that starts the server, then starts the client that sends the data to it and reads the result back. Its run method is:
sub run # ($labels)
{
my $labels = shift;
my ($port, $pid) = Triceps::X::SimpleServer::startServer(0, $labels);
my $sock = IO::Socket::INET->new(
Proto => "tcp",
PeerAddr => "localhost",
PeerPort => $port,
) or confess "socket failed: $!";
while(<STDIN>) {
$sock->print($_);
$sock->flush();
}
$sock->print("exit,OP_INSERT\n");
$sock->flush();
$sock->shutdown(1); # SHUT_WR
while(<$sock>) {
print($_);
}
waitpid($pid, 0);
}
It's really intended only for the very small examples that fit into the TCP buffer, since it sends the whole input before it starts reading the output.
The interesting server things happen inside startServer() which now also stayed almost the same but became a part of a module. The "almost the same" part is about the server loop being able to dispatch not only to the labels but also to the arbitrary Perl functions, citing from the example:
$dispatch{"query"} = sub { $tql->query(@_); };
$dispatch{"exit"} = \&Triceps::X::SimpleServer::exitFunc;
It recognizes automatically whether the entry in the dispatch table is a Label or a function, and handles them appropriately. In the server it's implemented with:
...
my $label = $labels->{$lname};
if (defined $label) {
if (ref($label) eq 'CODE') {
&$label($line);
} else {
my $unit = $label->getUnit();
confess "label '$lname' received from client $id has been cleared"
unless defined $unit;
eval {
$unit->makeArrayCall($label, @data);
$unit->drainFrame();
};
warn "input data error: $@\nfrom data: $line\n" if $@;
}
} else {
warn "unknown label '$lname' received from client $id: $line "
}
...
And the exitFunc() method is another way to trigger the server exit, instead of makeExitLabel():
sub exitFunc # ($line)
{
$srv_exit = 1;
}
As you can see, the dispatched functions receive the whole argument line as the client had sent it, including the label name, rather than having it split by commas. The functions can then do the text parsing in their own way, which comes real handy for TQL. It's convenient for the exit function too, as now there is no need to send the opcode with the "exit" (although X::DumbClient::run() still does send the opcode, to be compatible with the exit label approach, and the extra information doesn't hurt the exit function).
And now, the TQL definition. The TQL object gets created with the definition of a table, and then the TQL handler function shown above calls the method query() on it:
# The information about tables, for querying.
my $tql = Triceps::X::Tql->new(
name => "tql",
tables => [
$tWindow,
$tSymbol,
],
);
There are multiple ways to create the Tql objects. By default the option "tables" lists all the queryable tables, and their "natural" names will be used in the queries. It's possible to specify the names explicitly as well:
my $tql = Triceps::X::Tql->new(
name => "tql",
tables => [
$tWindow,
$tSymbol,
$tWindow,
$tSymbol,
],
tableNames => [
"window",
"symbol",
$tWindow->getName(),
$tSymbol->getName(),
],
);
This version defines each table under two synonymous names. It's also possible to create a Tql object without tables, and add tables to it later as they are created:
my $tql = Triceps::X::Tql->new(name => "tql");
$tql->addNamedTable(
window => $tWindow,
symbol => $tSymbol,
);
# add 2nd time, with different names
$tql->addTable(
$tWindow,
$tSymbol,
);
$tql->initialize();
The tables can be added with explicit names or with "natural" names. After all the tables are added, the Tql object has to be initialized. The two ways of creation are mutually exclusive: if the option "tables" is used, the object will be initialized right away in the constructor. If it's not used, the explicit initialization has to be done later. The methods addTable() and addNamedTable() can not be used on an initialized table, and query() can not be used on an uninitialized table.
# The basic table type to be used for querying.
# Represents the trades reports.
our $rtTrade = Triceps::RowType->new(
id => "int32", # trade unique id
symbol => "string", # symbol traded
price => "float64",
size => "float64", # number of shares traded
) or confess "$!";
our $ttWindow = Triceps::TableType->new($rtTrade)
->addSubIndex("bySymbol",
Triceps::SimpleOrderedIndex->new(symbol => "ASC")
->addSubIndex("last2",
Triceps::IndexType->newFifo(limit => 2)
)
)
or confess "$!";
$ttWindow->initialize() or confess "$!";
# Represents the static information about a company.
our $rtSymbol = Triceps::RowType->new(
symbol => "string", # symbol name
name => "string", # the official company name
eps => "float64", # last quarter earnings per share
) or confess "$!";
our $ttSymbol = Triceps::TableType->new($rtSymbol)
->addSubIndex("bySymbol",
Triceps::IndexType->newHashed(key => [ "symbol" ])
)
or confess "$!";
$ttSymbol->initialize() or confess "$!";
my $uTrades = Triceps::Unit->new("uTrades");
my $tWindow = $uTrades->makeTable($ttWindow, "EM_CALL", "tWindow")
or confess "$!";
my $tSymbol = $uTrades->makeTable($ttSymbol, "EM_CALL", "tSymbol")
or confess "$!";
# The information about tables, for querying.
my $tql = Triceps::X::Tql->new(
name => "tql",
tables => [
$tWindow,
$tSymbol,
],
);
my %dispatch;
$dispatch{$tWindow->getName()} = $tWindow->getInputLabel();
$dispatch{$tSymbol->getName()} = $tSymbol->getInputLabel();
$dispatch{"query"} = sub { $tql->query(@_); };
$dispatch{"exit"} = \&Triceps::X::SimpleServer::exitFunc;
Triceps::X::DumbClient::run(\%dispatch);
It's very much like the example shown before in the section 7.8 "Main loop with a socket", with a few differences. Obviously, Tql has been added, and we'll get to that part just in a moment. But the other differences are centered around the way the server and client code has been restructured.
The Triceps::X::DumbClient is a module for testing that starts the server, then starts the client that sends the data to it and reads the result back. Its run method is:
sub run # ($labels)
{
my $labels = shift;
my ($port, $pid) = Triceps::X::SimpleServer::startServer(0, $labels);
my $sock = IO::Socket::INET->new(
Proto => "tcp",
PeerAddr => "localhost",
PeerPort => $port,
) or confess "socket failed: $!";
while(<STDIN>) {
$sock->print($_);
$sock->flush();
}
$sock->print("exit,OP_INSERT\n");
$sock->flush();
$sock->shutdown(1); # SHUT_WR
while(<$sock>) {
print($_);
}
waitpid($pid, 0);
}
It's really intended only for the very small examples that fit into the TCP buffer, since it sends the whole input before it starts reading the output.
The interesting server things happen inside startServer() which now also stayed almost the same but became a part of a module. The "almost the same" part is about the server loop being able to dispatch not only to the labels but also to the arbitrary Perl functions, citing from the example:
$dispatch{"query"} = sub { $tql->query(@_); };
$dispatch{"exit"} = \&Triceps::X::SimpleServer::exitFunc;
It recognizes automatically whether the entry in the dispatch table is a Label or a function, and handles them appropriately. In the server it's implemented with:
...
my $label = $labels->{$lname};
if (defined $label) {
if (ref($label) eq 'CODE') {
&$label($line);
} else {
my $unit = $label->getUnit();
confess "label '$lname' received from client $id has been cleared"
unless defined $unit;
eval {
$unit->makeArrayCall($label, @data);
$unit->drainFrame();
};
warn "input data error: $@\nfrom data: $line\n" if $@;
}
} else {
warn "unknown label '$lname' received from client $id: $line "
}
...
And the exitFunc() method is another way to trigger the server exit, instead of makeExitLabel():
sub exitFunc # ($line)
{
$srv_exit = 1;
}
As you can see, the dispatched functions receive the whole argument line as the client had sent it, including the label name, rather than having it split by commas. The functions can then do the text parsing in their own way, which comes real handy for TQL. It's convenient for the exit function too, as now there is no need to send the opcode with the "exit" (although X::DumbClient::run() still does send the opcode, to be compatible with the exit label approach, and the extra information doesn't hurt the exit function).
And now, the TQL definition. The TQL object gets created with the definition of a table, and then the TQL handler function shown above calls the method query() on it:
# The information about tables, for querying.
my $tql = Triceps::X::Tql->new(
name => "tql",
tables => [
$tWindow,
$tSymbol,
],
);
There are multiple ways to create the Tql objects. By default the option "tables" lists all the queryable tables, and their "natural" names will be used in the queries. It's possible to specify the names explicitly as well:
my $tql = Triceps::X::Tql->new(
name => "tql",
tables => [
$tWindow,
$tSymbol,
$tWindow,
$tSymbol,
],
tableNames => [
"window",
"symbol",
$tWindow->getName(),
$tSymbol->getName(),
],
);
This version defines each table under two synonymous names. It's also possible to create a Tql object without tables, and add tables to it later as they are created:
my $tql = Triceps::X::Tql->new(name => "tql");
$tql->addNamedTable(
window => $tWindow,
symbol => $tSymbol,
);
# add 2nd time, with different names
$tql->addTable(
$tWindow,
$tSymbol,
);
$tql->initialize();
The tables can be added with explicit names or with "natural" names. After all the tables are added, the Tql object has to be initialized. The two ways of creation are mutually exclusive: if the option "tables" is used, the object will be initialized right away in the constructor. If it's not used, the explicit initialization has to be done later. The methods addTable() and addNamedTable() can not be used on an initialized table, and query() can not be used on an uninitialized table.
Sunday, November 25, 2012
TQL: the Trivial Query Language
In the Developer's Guide section 7.8. "Main loop with a socket" I've been showing the execution of the simple queries. I've wanted to use the queries to demonstrate a feature of the streaming functions, so I've substantially extended that example.
Now the query example has grown to have its own language, TQL. You can think of it as a Trivial Query Language or Triceps Query Language. It's trivial, and so far it's of only an example quality, but it's extensible and it already can do some interesting things.
Why not SQL, after all, there are multiple parser building tools available in Perl? Partially, because I wanted to keep it trivial and to avoid introducing extra dependencies, especially just for the examples. Partially, because I don't like SQL. I think that the queries can be expressed much more naturally in the form of shell-like pipelines. Back at DB when I wrote a simple toolkit for querying and comparison of the CSV files (yeah, I didn't find the DBD::CSV module), I've used a pipeline semantics and it worked pretty well. It also did things that are quite difficult with SQL, like mass renaming and reordering of fields, and diffing. Although TQL is not a descendant of the language I've used in that query tool, it is a further development of the pipeline idea.
Syntactically, TQL is very simple: its query is a represented as a nested list, similar to Tcl (or if you like Lisp better, you can think that it's similar to Lisp but with different parentheses). A list is surrounded by curly braces "{}". The elements of a list are either other lists or words, consisting of non-space characters.
{word1 {word21 word22} word3}
Unlike Tcl, there are no quotes in the TQL syntax, the quote characters are just the normal word characters. If you want to include spaces into a word, you use the curly braces instead of the quotes.
{ this is a {brace-enquoted} string with spaces and nested braces }
Note that the spaces inside a list are used as delimiters and thrown away but within a brace-quoted word-string they are significant. How do you know, which way they will be treated in a particular case? It all depends on what is expected in this case. If the command expects a string as an argument, it will treat it as a string. If the command expects a list as an argument, it will treat it as a list.
What if you need to include an unbalanced brace character inside a string? Escape it with a backslash, "\{". The other usual Perl backslash sequences work too (though in the future TQL may get separated from Perl and then only the C sequences will work, that is to be seen). Any non-alphanumeric characters (including spaces) can be prepended with a backslash too. An important point is that when you build the lists, unlike shell, and like Tcl, you do the backslash escaping only once, when accepting a raw string. After that you can include into the lists of any depth without any extra escapes (and you must not add any extra escapes in the lists).
Unlike shell, you can't combine a single string out of the quoted and unquoted parts. Instead the quoting braces work as implicit separators. For example, if you specify a list as {a{b}c d}, you don't get two strings "abc" and "d", you get four strings "a", "b", "c", "d".
A TQL query is a list that represents a pipeline. Each element of the list is a command. The first command reads the data from a table, and the following commands perform transformations on that data. For example:
{read table tWindow} {project fields {symbol price}} {print tokenized 0}
If the print command is missing at the end of the pipeline, it will be added implicitly, with the default arguments: {print}.
The arguments of each TQL command are always in the option name-value format, very much like the Perl constructors of many Triceps objects. There aren't any arguments in TQL that go by themselves without an option name.
So for example the command "read" above has the option "table" with value "tWindow". The command "project" has an option "fields" with a list value of two elements. In this case the elements are simple words and don't need the further quoting. But the extra quoting won't hurt. Say, if you wanted to rename the field "price" to "trade_price", you use the Triceps::Fields::filter() syntax for it, and even though the format doesn't contain any spaces and can be still used just as a word, it looks nicer with the extra braces:
{project fields {symbol {price/trade_price} }}
I'm sure that the list of commands and their options will expand and change over time. So far the supported commands are:
read
Defines a table to read from and starts the command pipeline.
Options:
table - name of the table to read from.
project
Projects (and possibly renames) a subset of fields in the current pipeline.
Options:
fields - an array of field definitions in the syntax of Triceps::Fields::filter() (same as in the joins).
print
The last command of the pipeline, which prints the results. If not used explicitly, the query adds this command implicitly at the end of the pipeline, with the default options.
Options:
tokenized (optional) - Flag: print in the name-value format, as in Row::printP(). Otherwise prints only the values in the CSV format. (default: 1)
join
Joins the current pipeline with another table.This is functionally similar to LookupJoin, although the options are closer to JoinTwo.
Options:
table - name of the table to join with. The current pipeline is considered the "left side", the table the "right side". The duplicate key fields on the right side are always excluded from the result, like JoinTwo option (fieldsUniqKey => "left").
rightIdxPath - path name of the table's index on which to join. At the moment there is no way to join without knowing the name of the index. (As usual, the path is an array of nested names).
by (semi-optional) - the join equality condition specified as pairs of fields. Similarly to JoinTwo, it's a single-level array with the fields logically paired:{leftFld1 rightFld1 leftFld2 rightFld2 ... }. Options "by" and "byLeft" are mutually exclusive, and one of them must be present.
byLeft (semi-optional) - the join equality condition specified as a transformation on the left-side field set in the syntax of Triceps::Fields::filter(), with an implicit element {!.*} added at the end. Options "by" and "byLeft" are mutually exclusive, and one of them must be present.
leftFields (optional) - the list of patterns for the left-side fields to pass through and possibly rename, in the syntax of Triceps::Fields::filter(). (default: pass all, with the same name)
rightFields (optional) - the list of patterns for the right-side fields to pass through and possibly rename, in the syntax of Triceps::Fields::filter(). The key fields get implicitly removed before. (default: pass all, with the same name)
type (optional) - type of the join, "inner" or "left". (default: "inner")
where
Filters/selects the rows.
Options:
istrue - a Perl expression, the condition for the rows to pass through. The particularly dangerous constructions are not allowed in the expression, including the loops and the general function calls. The fields of the row are referred to as $%field, these references get translated before the expression is compiled.
Here are some examples of the Tql queries, with results produced from the output of the code examples I'll show in a moment.
> query,{read table tSymbol}
lb1read OP_INSERT symbol="AAA" name="Absolute Auto Analytics Inc" eps="0.5"
+EOD,OP_NOP,lb1read
Reads the stock symbol information table and prints it in the default tokenized format. The result format is a bit messy for now, a mix of tokenized and CSV data. In the previous examples in the chapter 7 I've been marking the end-of-data either by a row with opcode OP_NOP or not marking it at all. For the TQL queries I've decided to try out a different approach: send a CSV row on the pseudo-label "+EOD" with the value equal to the name of the label that has been completed. The labels with names starting with "+" are special in this convention, they represent some kind of metadata.
The name "lb1read" in the result rows is coming from an auto-generated label name in TQL. It will probably become less random-looking in the future, but for now I haven't yet figured out the best way to to it.
> query,{read table tWindow} {project fields {symbol price}}
lb2project OP_INSERT symbol="AAA" price="20"
lb2project OP_INSERT symbol="AAA" price="30"
+EOD,OP_NOP,lb2project
Reads the trade window rows and projects the fields "symbol" and "price" from them.
> query,{read table tWindow} {project fields {symbol price}} {print tokenized 0}
lb2project,OP_INSERT,AAA,20
lb2project,OP_INSERT,AAA,30
+EOD,OP_NOP,lb2project
The same, only explicitly prints the data in the CSV format.
> query,{read table tWindow} {where istrue {$%price == 20}}
lb2where OP_INSERT id="3" symbol="AAA" price="20" size="20"
+EOD,OP_NOP,lb2where
Selects the trade window row with price equal to 20.
> query,{read table tWindow} {join table tSymbol rightIdxPath bySymbol byLeft {symbol}}
join2.out OP_INSERT id="3" symbol="AAA" price="20" size="20" name="Absolute Auto Analytics Inc" eps="0.5"
join2.out OP_INSERT id="5" symbol="AAA" price="30" size="30" name="Absolute Auto Analytics Inc" eps="0.5"
+EOD,OP_NOP,join2.out
Reads the trade window and enriches it by joining with the symbol information.
A nice feature of TQL is that it allows to combine the operations in the pipeline in any order, repeated any number of times. For example, you can read a table, filter it, join with another table, filter again, join with the third table, filter again and so on. SQL in the same situation has to resort to specially named clauses, for example WHERE filters before grouping and HAVING filters after grouping.
Of course, a typical smart SQL compiler would determine the earliest application point for each WHERE sub-expression and build a similar pipeline. But TQL allows to keep the compiler trivial, following the explicit pipelining in the query. And nothing really prevents a smart TQL compiler either, it could as well analyze, split and reorder the pipeline stages.
Now the query example has grown to have its own language, TQL. You can think of it as a Trivial Query Language or Triceps Query Language. It's trivial, and so far it's of only an example quality, but it's extensible and it already can do some interesting things.
Why not SQL, after all, there are multiple parser building tools available in Perl? Partially, because I wanted to keep it trivial and to avoid introducing extra dependencies, especially just for the examples. Partially, because I don't like SQL. I think that the queries can be expressed much more naturally in the form of shell-like pipelines. Back at DB when I wrote a simple toolkit for querying and comparison of the CSV files (yeah, I didn't find the DBD::CSV module), I've used a pipeline semantics and it worked pretty well. It also did things that are quite difficult with SQL, like mass renaming and reordering of fields, and diffing. Although TQL is not a descendant of the language I've used in that query tool, it is a further development of the pipeline idea.
Syntactically, TQL is very simple: its query is a represented as a nested list, similar to Tcl (or if you like Lisp better, you can think that it's similar to Lisp but with different parentheses). A list is surrounded by curly braces "{}". The elements of a list are either other lists or words, consisting of non-space characters.
{word1 {word21 word22} word3}
Unlike Tcl, there are no quotes in the TQL syntax, the quote characters are just the normal word characters. If you want to include spaces into a word, you use the curly braces instead of the quotes.
{ this is a {brace-enquoted} string with spaces and nested braces }
Note that the spaces inside a list are used as delimiters and thrown away but within a brace-quoted word-string they are significant. How do you know, which way they will be treated in a particular case? It all depends on what is expected in this case. If the command expects a string as an argument, it will treat it as a string. If the command expects a list as an argument, it will treat it as a list.
What if you need to include an unbalanced brace character inside a string? Escape it with a backslash, "\{". The other usual Perl backslash sequences work too (though in the future TQL may get separated from Perl and then only the C sequences will work, that is to be seen). Any non-alphanumeric characters (including spaces) can be prepended with a backslash too. An important point is that when you build the lists, unlike shell, and like Tcl, you do the backslash escaping only once, when accepting a raw string. After that you can include into the lists of any depth without any extra escapes (and you must not add any extra escapes in the lists).
Unlike shell, you can't combine a single string out of the quoted and unquoted parts. Instead the quoting braces work as implicit separators. For example, if you specify a list as {a{b}c d}, you don't get two strings "abc" and "d", you get four strings "a", "b", "c", "d".
A TQL query is a list that represents a pipeline. Each element of the list is a command. The first command reads the data from a table, and the following commands perform transformations on that data. For example:
{read table tWindow} {project fields {symbol price}} {print tokenized 0}
If the print command is missing at the end of the pipeline, it will be added implicitly, with the default arguments: {print}.
The arguments of each TQL command are always in the option name-value format, very much like the Perl constructors of many Triceps objects. There aren't any arguments in TQL that go by themselves without an option name.
So for example the command "read" above has the option "table" with value "tWindow". The command "project" has an option "fields" with a list value of two elements. In this case the elements are simple words and don't need the further quoting. But the extra quoting won't hurt. Say, if you wanted to rename the field "price" to "trade_price", you use the Triceps::Fields::filter() syntax for it, and even though the format doesn't contain any spaces and can be still used just as a word, it looks nicer with the extra braces:
{project fields {symbol {price/trade_price} }}
I'm sure that the list of commands and their options will expand and change over time. So far the supported commands are:
read
Defines a table to read from and starts the command pipeline.
Options:
table - name of the table to read from.
project
Projects (and possibly renames) a subset of fields in the current pipeline.
Options:
fields - an array of field definitions in the syntax of Triceps::Fields::filter() (same as in the joins).
The last command of the pipeline, which prints the results. If not used explicitly, the query adds this command implicitly at the end of the pipeline, with the default options.
Options:
tokenized (optional) - Flag: print in the name-value format, as in Row::printP(). Otherwise prints only the values in the CSV format. (default: 1)
join
Joins the current pipeline with another table.This is functionally similar to LookupJoin, although the options are closer to JoinTwo.
Options:
table - name of the table to join with. The current pipeline is considered the "left side", the table the "right side". The duplicate key fields on the right side are always excluded from the result, like JoinTwo option (fieldsUniqKey => "left").
rightIdxPath - path name of the table's index on which to join. At the moment there is no way to join without knowing the name of the index. (As usual, the path is an array of nested names).
by (semi-optional) - the join equality condition specified as pairs of fields. Similarly to JoinTwo, it's a single-level array with the fields logically paired:{leftFld1 rightFld1 leftFld2 rightFld2 ... }. Options "by" and "byLeft" are mutually exclusive, and one of them must be present.
byLeft (semi-optional) - the join equality condition specified as a transformation on the left-side field set in the syntax of Triceps::Fields::filter(), with an implicit element {!.*} added at the end. Options "by" and "byLeft" are mutually exclusive, and one of them must be present.
leftFields (optional) - the list of patterns for the left-side fields to pass through and possibly rename, in the syntax of Triceps::Fields::filter(). (default: pass all, with the same name)
rightFields (optional) - the list of patterns for the right-side fields to pass through and possibly rename, in the syntax of Triceps::Fields::filter(). The key fields get implicitly removed before. (default: pass all, with the same name)
type (optional) - type of the join, "inner" or "left". (default: "inner")
where
Filters/selects the rows.
Options:
istrue - a Perl expression, the condition for the rows to pass through. The particularly dangerous constructions are not allowed in the expression, including the loops and the general function calls. The fields of the row are referred to as $%field, these references get translated before the expression is compiled.
Here are some examples of the Tql queries, with results produced from the output of the code examples I'll show in a moment.
> query,{read table tSymbol}
lb1read OP_INSERT symbol="AAA" name="Absolute Auto Analytics Inc" eps="0.5"
+EOD,OP_NOP,lb1read
Reads the stock symbol information table and prints it in the default tokenized format. The result format is a bit messy for now, a mix of tokenized and CSV data. In the previous examples in the chapter 7 I've been marking the end-of-data either by a row with opcode OP_NOP or not marking it at all. For the TQL queries I've decided to try out a different approach: send a CSV row on the pseudo-label "+EOD" with the value equal to the name of the label that has been completed. The labels with names starting with "+" are special in this convention, they represent some kind of metadata.
The name "lb1read" in the result rows is coming from an auto-generated label name in TQL. It will probably become less random-looking in the future, but for now I haven't yet figured out the best way to to it.
> query,{read table tWindow} {project fields {symbol price}}
lb2project OP_INSERT symbol="AAA" price="20"
lb2project OP_INSERT symbol="AAA" price="30"
+EOD,OP_NOP,lb2project
Reads the trade window rows and projects the fields "symbol" and "price" from them.
> query,{read table tWindow} {project fields {symbol price}} {print tokenized 0}
lb2project,OP_INSERT,AAA,20
lb2project,OP_INSERT,AAA,30
+EOD,OP_NOP,lb2project
The same, only explicitly prints the data in the CSV format.
> query,{read table tWindow} {where istrue {$%price == 20}}
lb2where OP_INSERT id="3" symbol="AAA" price="20" size="20"
+EOD,OP_NOP,lb2where
Selects the trade window row with price equal to 20.
> query,{read table tWindow} {join table tSymbol rightIdxPath bySymbol byLeft {symbol}}
join2.out OP_INSERT id="3" symbol="AAA" price="20" size="20" name="Absolute Auto Analytics Inc" eps="0.5"
join2.out OP_INSERT id="5" symbol="AAA" price="30" size="30" name="Absolute Auto Analytics Inc" eps="0.5"
+EOD,OP_NOP,join2.out
Reads the trade window and enriches it by joining with the symbol information.
A nice feature of TQL is that it allows to combine the operations in the pipeline in any order, repeated any number of times. For example, you can read a table, filter it, join with another table, filter again, join with the third table, filter again and so on. SQL in the same situation has to resort to specially named clauses, for example WHERE filters before grouping and HAVING filters after grouping.
Of course, a typical smart SQL compiler would determine the earliest application point for each WHERE sub-expression and build a similar pipeline. But TQL allows to keep the compiler trivial, following the explicit pipelining in the query. And nothing really prevents a smart TQL compiler either, it could as well analyze, split and reorder the pipeline stages.
Subscribe to:
Posts (Atom)