TelegraphCQ v2.1

• Introduction

This is TelegraphCQ version 2.1, the latest release of TelegraphCQ by Telegraph Team at UC Berkeley. This release, as all the previous ones, is under the OpenBSD license.

TelegraphCQ is a system for processing long-running continuous queries over data streams. It is implemented as a significant modification to the PostgreSQL open-source database system. Specifically, TelegraphCQ v2.1 is based on the PostgreSQL 7.3.2 code-base.

What's new

• Better stability: Extensive testing. Lots of bugs have been squashed
  
• Better performance: System has been load-tested, profiled and optimized to a large extent.
  
• New window semantics: You can now specify a sliding, hopping or jumping time window over streams using the window clause: [RANGE BY '' SLIDE BY '' START AT '']. In addition, a special aggregate, wtime(*), returns the latest timestamp in the current time window.
  
• SQL:1999-like WITH syntax and recursive queries.
• Heartbeats: We've added support for heartbeats using punctuation tuples.
  
• Summarize on drop (Data Triage): Tuples dropped on stream overload can now be summarized.
• Flux: Infrastructure for a  dataflow operator that encapsulates adaptive state partioning
  and data flow routing.
  

• Installation
  

• Pre-requisites

• Platforms

TelegraphCQ version 2.1 has been built and tested extensively on Linux Fedora Core 1.

• Additional libraries

TelegraphCQ requires that these packages be installed on your system. ${HOME}/tcq-tools is a convenient place to install them, since that's where TelegraphCQ's configure-opts script looks for these libraries.


• mm

mm is a shared memory allocation package developed by OSSP, the Open Source Software project. Extract the latest tar file and follow the installation instructions
found in path-to-mm/INSTALL, which are summarized below:

$ ./configure --prefix=${HOME}/tcq-tools $ make $ make test $ make install

• Judy
  

Judy is a high-performance C-library for implementing sparse arrays. We use a version that has been patched specifically to work with TelegraphCQ. Again, extract the tar.gz file and follow the installation instructions in path-to-judy/INSTALL document, as summarized below:

$ ./configure --prefix=${HOME}/tcq-tools $ make $ make check $ make install

• Building TelegraphCQ

Run configure to set up the build parameters for your system. For those unfamiliar with configure, it’s part of the GNU Build system. configure-tcq is a convenient script that runs configure with many commonly used flags. configure-tcq looks in ${HOME}/tcq-tools for the additional libraries mentioned above. If they have been installed elsewhere, modify configure-tcq to point it to the correct location for these libraries. Before running configure-tcq, the environment variable PGRSHOME should be set to point to the TelegraphCQ installation directory. completes successfully, run make as shown below:

$ ./configure-tcq $ make $ make install

• Getting Started
  

TelegraphCQ is built on the PostgreSQL code-base. Hence, you can start the postmaster either in PostgreSQL mode, or in TelegraphCQ mode. To start using TelegraphCQ, we follow the steps below:


• Create a new database cluster
  

$ initdb --no-locale -D ${PGDATA}

initdb initializes the data directories, the shared catalogs and performs other administrivia like, locale and character set encoding. As an alternative to the -D option, you can set PGDATA to point to the data directory on the file system. By passing it the --no-locale flag, initdb inherits the locale from it's environment.


• Create a new database
  

$ pg_ctl start -D ${PGDATA} -l <logfile-name> $ createdb ${DBNAME} $ pg_ctl stop -D ${PGDATA} -l <logfile-name>

createdb creates a new PostgreSQL database. pg_ctl start|stop|restart are commands used to start/stop or restart a PostgreSQL server. logfile-name specifies the file to which all status messages are logged.


• Start PostgreSQL in TelegraphCQ mode
  

$ pg_ctl start -D ${PGDATA} -l logfile-name -o " -t database-name -u \ user-name -G -i -Q 64 -d 1"

pg_ctl passes all the options within the -o flag to the postmaster, which is the PostgreSQL multi-user database server.
-t is the database to connect to.
-d Sets the debug level. The higher this value, the more debugging output is written to the log.
-i Allows clients to connect using TCP/IP sockets.
-G Starts the TelegraphCQ executor in eddy mode.
-Q Sets the maximum number of result queues that can be created.


• Start the client
  

$ psql -C database-name

psql is an interactive client to PostgreSQL. It allows you to pose queries and see the results.
-C Submits SELECT queries using cursors. Since data streams are potentially infinite, we can't wait
for all the results to be produced. Instead, we use a cursor to iterate over the results. We can control the size and duration of the results using two additional flags.
-I n Will fetch results in batches of n rows. The default value of n is 1.
-i m Will cancel the query after m rows have been fetched. The default value of m is infinity.


• Define a stream
Define a data stream using the CREATE STREAM command. It's syntax is similar to the
CREATE TABLE command. Let's assume that we connected to a database named sample in the previous step. You'll get the psql prompt when the client connects successfully to the database.


Welcome to psql 0.2, the PostgreSQL interactive terminal. Type: \copyright for distribution terms \h for help with SQL commands \? for help on internal slash commands \g or terminate with semicolon to execute query \q to quit starting in CQ Query mode. All queries will be submitted using cursors. sample=# CREATE SCHEMA traffic; sample=# CREATE STREAM traffic.measurements (stationid INT, speed REAL, tcqtime TIMESTAMP TIMESTAMPCOLUMN ) TYPE ARCHIVED;

tcqtime is the default timestamp column which obeys the TIMESTAMPCOLUMN constraint. It specifies the creation time of a tuple, which are assumed to be monotonically increasing. Tuples entering TelegraphCQ can either be externally time-stamped, or time-stamped by the data source. If they're time-stamped by a TelegraphCQ data source like the CSV Wrapper, they will be in monotonically increasing order.

Data streams in TelegraphCQ can be of two types, archived or unarchived. Tuples streamed into archived streams are copied to disk by TelegraphCQ, while data streamed to unarchived queries are discarded when the query is cancelled. Currently streams, once archived, can't be queried in real-time. But, you can restart the server in PostgreSQL mode and run queries over the entire archived stream.


• Associate a wrapper with the stream

sample=# ALTER STREAM traffic.measurements ADD WRAPPER csvwrapper;

Wrappers are user-defined data acquistion functions in TelegraphCQ. ALTER STREAM associates the stream with a CSV wrapper.  A CSV wrapper, that can handle input in the Comma Separated Value file format, comes bundled with TelegraphCQ.


• A simple windowed query

sample=# SELECT dst, COUNT(*), wtime(*) AS c FROM network.tcpdump AS st [RANGE BY '5 seconds' SLIDE BY '1 second' START AT '2003-06-06 18:50:20'] GROUP BY dst;

The window clause [RANGE BY ... SLIDE BY ... START AT ...] defines a window over potentially infinite input data stream, thereby placing a bound on it and enabling us to return results without waiting for the entire data stream to be processed. The RANGE parameter defines the size of the window in number of seconds, and the SLIDE parameter defines the interval after which this window will be re-calculated, again in number of seconds. The START AT parameter defines the time at which the first window begins.


• Send input data to stream

$ cat tcpdump.log | source.pl localhost 5533 csvwrapper,network.tcpdump

tcpdump.log is a CSV file containing a dump of TCP message headers, produced by the UNIX command tcpdump, a snippet of which is given below:

128.32.37.185,32797,64.174.7.0,80,S,06/06/2003 18:50:20.856709 128.32.37.185,32797,64.174.7.1,80,S,06/06/2003 18:50:21.856709 128.32.37.185,32797,64.174.7.2,80,S,06/06/2003 18:50:22.856709 128.32.37.185,32797,64.174.7.3,80,S,06/06/2003 18:50:23.856709 128.32.37.185,32797,64.174.7.4,80,S,06/06/2003 18:50:24.856709 128.32.37.185,32797,64.174.7.5,80,S,06/06/2003 18:50:25.856709

source.pl is a Perl script connects and feeds data to the wrapper clearinghouse, which is a dedicated process that listens to connection requests from new data sources. By default, it listens on port 5533. In addition, we also specify the wrapper type and the schema of the stream we're providing input to. wtime is a special aggregate over windows which returns the  latest timestamp of all tuples in that window. A snippet of the query result is given below:


dst | count | c ---------------+-------+--------------------- 64.174.7.0/32 | 1 | 2003-06-06 18:50:25 64.174.7.1/32 | 1 | 2003-06-06 18:50:25 64.174.7.2/32 | 1 | 2003-06-06 18:50:25 64.174.7.3/32 | 1 | 2003-06-06 18:50:25 64.174.7.4/32 | 1 | 2003-06-06 18:50:25 64.174.7.1/32 | 1 | 2003-06-06 18:50:26 64.174.7.2/32 | 1 | 2003-06-06 18:50:26



• More complex queries
  Now that we've seen a simple windowed query, let's look at how we'd specify joins and recursive queries in TelegraphCQ.
  
  • Joins
  Joins on streams are similar to those on relations, except that, in order to bound the join state each stream should have window specifications.

  SELECT R.i, R.j, count(*) FROM R [RANGE BY 't1 seconds' SLIDE BY 't2 seconds'], S [RANGE BY 't3 seconds' SLIDE BY 't4 seconds'] WHERE R.k = S.k GROUP BY R.i, R.j HAVING R.j > C;

  • Recursive queries, the WITH clause
  You can write recursive queries using the SQL:1999-like WITH syntax.
  

  WITH StreamOne AS ( SELECT R.i, sum(R.j) as sum, wtime(*) FROM R [RANGE BY 't1 seconds' SLIDE BY 't2 seconds'] ) StreamTwo AS ( SELECT S.k, sum(S.l) as sum, wtime(*) FROM S [RANGE BY 't3 seconds' SLIDE BY 't4 seconds'] ) (SELECT * FROM StreamOne S1, StreamTwo S2 WHERE S1.i = S2.k);

  • Windowed aggregates
  Like joins, aggregates on streams are computed on windows since we need to bound the aggregate state.

  SELECT SUM(R.i), AVG(R.j), COUNT(*), wtime(*) FROM R [RANGE BY 't1 seconds' SLIDE BY 't2 seconds'];

• The Sanity Test suites
  
The sanity test suite, found in src/test/examples/tcqsanity/ contains a comprehensive set of tests for TelegraphCQ's features. To ensure deterministic behavior, this test suite runs a diverse set of queries against canned data, such that the output can be compared automatically against expected outputs. src/test/examples/csv/streamsanity.pl has a set of timing-dependent queries whose outputs have to be visually inspected for correctness.

Both test suites contain good examples of complex, windowed queries on continuous data streams.

• TelegraphCQ architectural overview
  

For those familiar with PostgreSQL, both the TelegraphCQ front-end (TFE) and the TelegraphCQ back-end (TBE) are PostgreSQL backends. There are multiple TFEs (one per client connection) and one TBE.

As shown in the Architecture diagram below,  a separate PostgreSQL process (TFE) is forked for each client connection.  This process runs all queries that do not involve streams using the normal PostgreSQL executor.  When PostgreSQL receives a continuous query, the TFE parses and plans the query in shared memory. It uses the output of the PostgreSQL optimizer to construct a continuous query plan.  Next, the TFE passes the plan to the TelegraphCQ backend executor (TBE) using a shared memory queue. 

The TBE runs the TelegraphCQ eddy which merges all continuous query plans into one so that query processing may be shared amonst queries. The TBE receives the plan, and integrates it into the TelegraphCQ eddy.  Finally, the Eddy returns results to the appropriate TFE via shared memory result queues, one per query.








• The TelegraphCQ Executor: Sharing Explained

The TelegraphCQ executor is a eddy, a continuously adaptive query processing mechanism.  An eddy is a routing operator that contains a number of modules that perform work on behalf of queries, and a number of sources that provide input data.  The eddy obtains data from sources, determines which modules a particular tuple must visit before all processing for the tuple is complete.  After a tuple has visited all required modules, it is output to all relevant result queues by the eddy.    

The continuous query plan consists of a number of distinct modules:



Filters	These represent single-table qualifications in the query. They are always of the form column OP constant. These filter modules will be merged together by the eddy to form a grouped filter which can evaluate filter predicates for multiple queries simultaneously.
SteMs	The building blocks for continuous query joins. A SteM holds tuples for a particular base relation or stream, and can be probed efficiently to find data tuples which match a probe tuple.
Table or stream Scans	These scans are used to obtain data from TelegraphCQ streams or regular PostgreSQL relations.





• Data Acquisition Mechanisms in TelegraphCQ
  

TelegraphCQ acquires data from external sources into streams.  These sources must connect to the TelegraphCQ system and identify themselves prior to sending data. User-defined data acquisition functions called wrappers allow substantial flexibility as to how that data is processed prior to entry into the TelegraphCQ system and insertion into a stream.  

User-defined wrapper functions manage reading data from a networked source, processing it, and returning it to the system in the appropriate format.  These functions are declared and managed using the PostgreSQL user-defined function infrastructure.  A dedicated wrapper clearinghouse process accepts connections from new data providers,  calls the appropriate wrapper functions in response to data arrival or other events such as connection termination or after a specified period of inactivity, and routes data tuples  returned by the wrappers to the corresponding streams.


• Streams

A stream appears to the user as if it were any other PostgreSQL table. An unarchived stream is never backed by disk storage, and uses shared-memory queues to communicate between the wrapper clearinghouse and the executor.  An archived stream uses an append-only access method to pass tuples to the rest of the system.  

Streams must have at most one column named tcqtime and of type TIMESTAMP.  In addition this column must have the TIMESTAMPCOLUMN constraint defined on it.  TelegraphCQ uses this column as the timestamp for windowed query operations.


• Data Sources
  

TelegraphCQ supports acquiring data from two different types of sources. Push sources initiate a connection to TelegraphCQ, send some configuration information, and then send data rows to the system. Pull sources provide a way for TelegraphCQ to initiate requests for data from other data sources.

Data Acquisition in TelegraphCQ has more on how to write user-defined wrapper functions and create sources.
 
You can also integrate Web based data sources with TelegraphCQ.

• Introspective query processing: self-monitoring capability
  

TelegraphCQ also has some special streams that report information about the state of the system.


• Windows and Streams

Unlike normal relations, data streams typically do not end.  It is therefore useful to be able to define windows, or subsets of the streams, which participate in a query result.

Each stream has a special time attribute that TelegraphCQ uses as the tuple’s timestamp for windowed operations.  Windows on a stream are defined in terms of a time interval relative to the most recently arrived tuple. The timestamp for a join tuple is defined as the maximum stream timestamp contained in the composite tuple.  

The creation of a new tuple that has a stream timestamp greater than any seen so far marks the beginning of a new window for query operations.


• Running Queries Over Streams

Queries over data streams are run using the SQL SELECT statement.  Queries that contain streams do not support the full SELECT syntax, and also include extra RANGE BY, SLIDE BY and START AT predicates to specify the window size for stream operations.  

The modified SELECT statement has the following form:


SELECT <select_list>
FROM <relation_and_stream_list>
WHERE <predicate>
GROUP BY <group_by_expressions>
ORDER BY <order_by_expressions>;


The stream_list may contain one window expression ([RANGE BY ... SLIDE BY ... START AT...]) per stream in the query.  A window expression specifies the correlation name for the stream and the size of the sliding window expressed as a PostgreSQL interval datatype.

All other clauses in the SELECT statement behave like the PostgreSQL select statement with the following additional restrictions:


1. Windows may not be defined over PostgreSQL relations.
2. WHERE clause qualifications that join two streams may only involve columns, not column expressions or functions.
3. WHERE clause qualifications that filter tuples must be of the form column OP constant
4. The WHERE clause may contain ANDs, but not ORs.
5. GROUP BY and ORDER BY clauses may only appear if the query also contains a window. 



• Command reference
  

• CREATE STREAM
• DROP STREAM
• CREATE WRAPPER
• DROP WRAPPER
• ALTER STREAM
  
• SELECT
  

• Run-time configuration

TelegraphCQ are a number of parameters that can be tweaked in accordance with user needs. The more important ones are listed below:

• numResultQueues
Sets the maximum number of result queues that can be instantiated. It can be controlled at run-time by passing the -Q N flag (upto a maximum of 128) to the postmaster.
 
The following parameters can be found in src/include/telegraphcq/telegraphcqinit.h.


• TCQMAXQUERIES
Sets the maximum number of queries in the system.
• TCQMAXOPERATORS
Sets the maximum number of operators in the system.
• TCQMAXSOURCES
Sets the maximum number of sources in the system.
• TCQTSATTRNAME
Defines the name of the column that obeys TIMESTAMPCOLUMN constraint.

• Publications

For the curious, publications pertaining to the overall architecture of TelegraphCQ, the Fjord and Eddy operators, as well as State Modules (STeMs) are listed below:
• Sirish Chandrasekaran, Owen Cooper, Amol Deshpande, Michael J. Franklin, Joseph M. Hellerstein, Wei Hong, Sailesh Krishnamurthy, Samuel R. Madden, Vijayshankar Raman, Fred Reiss, and Mehul A. Shah. TelegraphCQ: Continuous Dataflow Processing for an Uncertain World. CIDR 2003. [PDF]
  
• Sailesh Krishnamurthy, Sirish Chandrasekaran, Owen Cooper, Amol Deshpande, Michael J. Franklin, Joseph M. Hellerstein, Wei Hong, Samuel R. Madden, Vijayshankar Raman, Fred Reiss, and Mehul A. Shah. TelegraphCQ: An Architectural Status Report. IEEE Data Engineering Bulletin, Vol 26(1), March 2003. [PDF]
• Vijayshankar Raman, Amol Deshpande, and Joseph M. Hellerstein. Using State Modules for Adaptive Query Processing. ICDE 2003. [PDF]
• Mehul A. Shah, Joseph M. Hellerstein, Sirish Chandrasekaran, and Michael J. Franklin. Flux: An Adaptive Partitioning Operator for Continuous Query Systems. ICDE 2003. [PDF]
• Samuel R. Madden, Mehul A. Shah, Joseph M. Hellerstein and Vijayshankar Raman. Continuously Adaptive Continuous Queries over Streams. ACM SIGMOD Conference, Madison, WI, June 2002. [PDF]
• Samuel R. Madden and Michael J. Franklin. Fjording the Stream: An Architecture for Queries over Streaming Sensor Data ICDE Conference, February, 2002, San Jose. [PDF]
• Joseph M. Hellerstein and Ron Avnur. Eddies: Continuously Adaptive Query Processing. In SIGMOD 2000. [PDF]



• Additional references

• postgresql.org is an invaluable site with a wealth of information, both on using PostgreSQL and developing on it.

• Examples of TelegraphCQ applications
  

Examples of projects that have used TelegraphCQ are given below. Please bear in mind that these projects used earlier versions of TelegraphCQ.
• Using Data Stream Management Systems for Traffic Analysis A Case Study, Thomas Plagemann, Vera Goebel, Andrea Bergamini, Giacomo Tolu, Guillaume Urvoy-Keller, Ernst W. Biersack. In the 5th Passive and Active Measurement workshop, Antibes Juan-les-Pins, France, April 2004.[PDF]
• A Flexible Architecture for Statistical Learning and Data Mining from System Log Streams, Wei Xu, Peter Bodik, David Patterson. In the workshop on Temporal Data Mining: Algorithms, Theory and Applications at The Fourth IEEE International Conference on Data Mining (ICDM'04), Brighton, UK, November 2004.[PDF]

• Appendixes

• Appendix A: The Wrapper Writer's Guide
• Appendix B: TelegraphCQ Window Semantics
• Appendix C: Load Shedding
• Appendix D: Heartbeats