The CREATE STREAM statement will create a stream definition which may be used in subsequent statements such as SELECT.

CREATE STREAM stream_name (col_name data_type [, ...])
    FROM [LOCAL] {FILE | NODE | SOURCE} input_spec
    [EVENT FORMAT format_spec [PROPERTIES (key = val, …)]]

CREATE STREAM stream_name AS select_statement

data_type ::= BOOLEAN | BINARY | BIGINT | INT | FLOAT | DOUBLE | PRECISE(int) | STRING | TIMESTAMP 
format_spec ::= 'delimited' | 'regex' | 'avro'
        

Section 6.3, “Data types and value ranges” describes the rtsql data types in greater detail.

input_spec is a 'single-quoted-string' identifying the filename / Flume logical node / Flume source specification to use as the input for this stream.

File names are Hadoop Path objects; they may specify the complete URI to a file, using any protocol permitted by the Hadoop common library. e.g.:

rtsql> CREATE STREAM foo (x STRING) FROM FILE
    -> 'hdfs://nn.example.com/user/aaron/foo.txt';
          

Unqualified file names are interpreted relative to the value of the fs.default.name configuration parameter. For example, if this were set to 'hdfs://nn.example.com', the following definition would be equivalent to the previous one:

rtsql> CREATE STREAM foo (x STRING) FROM FILE '/user/aaron/foo.txt';
          

Using the LOCAL keyword will cause the source definition to be interpreted relative to the local filesystem of the FlumeBase server. The following two statements are equivalent:

rtsql> CREATE STREAM foo (x STRING) FROM LOCAL FILE '/home/aaron/foo.txt';
rtsql> CREATE STREAM foo (x STRING) FROM FILE 'file:///home/aaron/foo.txt';
          

Note that if the FlumeBase server is on a different machine than the client, this will read from /home/aaron/foo.txt on the FlumeBase server -- not the client.

The EVENT FORMAT clause specifies how the bytes inside an event should be interpreted. By default, rtsql uses the delimited event format. Events are assumed to contain UTF-8 text representations of each field, separated by commas.

By specifying an EVENT FORMAT, you can choose which parser to apply to each event. The event format is specified as a 'quoted string'. The next few subsections define the available event formats.

You can further control the behavior of the event parser by specifying (key, value) pairs in the PROPERTIES section. The keys recognized are specific to each event format. Keys and values are both single-quoted strings.

  rtsql> CREATE STREAM x(a int, b int) FROM LOCAL FILE 'foo.txt'
      -> EVENT FORMAT 'delimited' PROPERTIES ('delimiter' = '|');
            

The DROP STREAM statement removes a stream definition created by CREATE STREAM.

DROP STREAM stream_name
        

When dropping a stream created in terms of a flow (CREATE STREAM AS SELECT), this will decommission the Flume logical node and drop the stream identifier, but will not cancel the flow itself. See Section 8.4, “Controlling flows” for more information on how to cancel the flow itself.

The SHOW STREAMS statement shows the definitions of all streams.

The SHOW FUNCTIONS statement shows the definitions of all functions which may be applied to expressions in a statement. The output of this command is a list of functions and their types. Types are written in the form ((input_types) -> output_type).

rtsql> SHOW FUNCTIONS;
length ((STRING) -> INT)
...
          

The length function may take a STRING or NULL value, and returns an INT (or NULL, if the input was NULL).

Some functions are polymorphic -- their input types are flexible, subject to certain constraints, and their output types may or may not match their input types. For example, the sum function can operate over any numeric type:

rtsql> SHOW FUNCTIONS;
sum ((var('a, constraints={TYPECLASS_NUMERIC})) -> var('a, constraints={TYPECLASS_NUMERIC}))
...
          

The input argument’s type is var('a, constraints={ TYPECLASS_NUMERIC}). This is a type variable with the name 'a (pronounced "alpha"), and can take any type subject to the constraint that it is in the typeclass "numeric" -- that is, it is one of INT, BIGINT, FLOAT, or DOUBLE. It is an error to take the sum of a STRING or BOOLEAN column.

The output argument is the same type variable "alpha;" whatever type is used for the input, will also be used as the output type. For more information on polymorphic types, see Section 6.3.10, “Polymorphic types and type classes”.

A complete list of functions included with rtsql is provided in Section 9, “Function Reference”.

The DESCRIBE statement shows the definition of a single object in rtsql:

DESCRIBE identifier        
        

This may be used to inspect a single stream, function, or other entity present in the symbol table.

The following statement displays the argument and return types for the length function:

rtsql> DESCRIBE length;
length ((STRING) -> INT)
          

The EXPLAIN statement shows the execution plan for an rtsql statement:

EXPLAIN statement        
        

This may be used to inspect the operation of any rtsql statement. The output of the command is a text description of how the statement was parsed (in a tree-based representation), followed by a control-flow graph of the steps applied in the runtime environment to satisfy the query.

rtsql> EXPLAIN SELECT x FROM foo;        
        

stream_reference ::= (stream_name | select_statement) [[AS] ref_name]
        

The stream_reference in a SELECT statement may literally identify a stream:

rtsql> CREATE STREAM foo (x string) FROM ...;
CREATE STREAM
rtsql> SELECT * FROM foo;
...
          

You may also qualify column names with their stream name:

rtsql> SELECT foo.x FROM foo;
          

And you may provide a reference name (ref_name) that is different than the stream name:

rtsql> SELECT v.x FROM verylongname AS v;
          

The AS keyword is optional. This is equivalent to:

rtsql> SELECT v.x FROM verylongname v;
          

A stream_reference may also be a nested SELECT statement.

rtsql> SELECT LENGTH(x) FROM (SELECT x FROM foo) AS f;
          

Each nested SELECT statement must be given a ref_name alias (f in the previous example). You do not need to qualify individual column names with the ref_name unless the column name would otherwise be ambiguous (e.g., if two sources are joined, and they each contain a column named x, then all references to x must be qualified with the source ref_name).

where_clause ::= WHERE bool_expr
        

A SELECT statement may filter some input events, and emit output events corresponding only to input events that match a boolean predicate.

rtsql> SELECT x FROM foo WHERE LENGTH(x) > 5;
          

This may be a compound boolean expression (using the AND and OR operators). rtsql does not support the IN or EXISTS operators. Subqueries are also not permitted in a WHERE clause.

join_clause ::= JOIN stream_reference ON join_expr OVER range_expr
        

A SELECT statement may correlate events from multiple sources and operate on their joined representation. In table-based SQL systems, any row of one table may be joined with any row of another table in a JOIN clause. Since FlumeBase operates over potentially infinite streams of data, this model would not scale. Instead, JOIN clauses require a window clause which defines the time-based boundaries within which a join may occur.

The only join expression supported is an equi-join; the join_expr must use the equality operator (=) to relate one field of each of the two joined streams.

The range_expr specifies the time range for the dependent (right) stream in which events may be joined to a given event of the primary (left) stream.

range_expr ::= RANGE INTERVAL expr time_scale PRECEDING
    | BETWEEN INTERVAL expr time_scale PRECEDING AND INTERVAL expr time_scale FOLLOWING

time_scale ::= SECONDS | MINUTES | HOURS | DAYS | WEEKS | MONTHS | YEARS
        

Consider the following example:

rtsql> SELECT * FROM f JOIN g ON f.x = g.y
    -> OVER RANGE INTERVAL 5 SECONDS PRECEDING;
          

This specifies that for each event seen in f (the primary stream), it may be joined with any events in g (the dependent stream) which occurred up to five seconds before the event in f.

The opposite time relation holds from the perspective of events in g: for each event in stream g, it may be joined with any events in f which occur up to five seconds later.

rtsql> SELECT * FROM f JOIN g ON f.x = g.y OVER
    -> RANGE BETWEEN INTERVAL 1 SECONDS PRECEDING
    -> AND INTERVAL 5 SECONDS FOLLOWING;
          

This example specifies that each event in f may be joined with any matching events in g which occured up to one second before, or five seconds after the event in f.

Only inner joins are supported at present. The INNER, OUTER, NATURAL, LEFT, RIGHT, and FULL keywords are not (yet) supported by rtsql.

Aggregate operators may be used in rtsql in a similar manner to ordinary SQL systems.

group_by_clause ::= GROUP BY col [, col...]
over_clause ::= OVER range_expr
having_clause ::= [ HAVING bool_expr ]
        

If an aggregate function (e.g., sum, count) is used in a SELECT statement, the statement must have an over_clause which defines the time window in which aggregate operators work. Aggregation over "the entire stream" is not supported.

The range_expr syntax is given in Section 6.2.3, “JOIN clauses”.

The following example provides a count of the number of events observed over a rolling five second window:

rtsql> SELECT COUNT(*) FROM foo OVER RANGE INTERVAL 5 SECONDS PRECEDING;
          

This may be further refined with a group_by_clause. For example:

rtsql> SELECT COUNT(*) FROM foo GROUP BY event_src
    -> OVER RANGE INTERVAL 5 SECONDS PRECEDING;
          

FlumeBase uses bucketing to support rolling time windows. By default, 100 buckets are used. So the previous two examples will support rolling counts with a "step" size of 50 milliseconds (5 seconds / 100 buckets).

You can specify the number of buckets by setting the flumebase.aggregation.buckets key in the session configuration (See Section 8.5, “Controlling the session configuration”). A larger number of buckets allows finer granularity in rolling windows, but may increase memory usage.

By default, windowed aggregation operators will emit output groups only when the corresponding input buckets contained data. For example, if data arrives at t=100 and t=150, and buckets are 10 ms wide, an output group will be emitted with timestamp t=100 and another one for t=150, but no intermediate counts or other aggregates will be generated for t=110, t=120, etc. This behavior can be configured by setting flumebase.aggregation.continuous.output in the session configuration. Setting this flag to true will cause the aggregation operator to emit output groups for all time intervals in which any data is available. This may prove more useful for building time-series graphs, etc, but less so for working with sporadic incoming data.

Note that by setting flumebase.aggregation.buckets to 1 and flumebase.aggregation.continuous.output to true, you can disable rolling windows, and instead divide the stream into discrete time-based groups.

This example gives a minute-by-minute summary of hits per minute from a web log:

rtsql> \set flumebase.aggregation.buckets=1;
rtsql> \set flumebase.aggregation.continuous.output=true;
rtsql> SELECT COUNT(*) as hits FROM httpd_log
    -> OVER RANGE INTERVAL 1 MINUTES PRECEDING;
          

Flume may deliver events out of order; FlumeBase tolerates improperly ordered events that arrive within the "slack interval" of 200 milliseconds of when they are expected. Events that arrive after the slack interval may be excluded from aggregate functions. The slack interval may be configured by setting flumebase.slack.time to a different integer number of milliseconds in the session configuration. (See Section 8.5, “Controlling the session configuration”.)

The set of aggregate functions available in rtsql are described in Section 9.2, “Aggregate functions”.

rtsql does not support full SQL:2003 windowed operators; a single range for the entire select statement is applied by the over_clause to all aggregate operators. This may change in a future version of FlumeBase.

having_clause ::= [ HAVING bool_expr ]         
          

rtsql> SELECT event_src, COUNT(event_src) AS cnt FROM foo
    -> GROUP BY event_src
    -> OVER RANGE INTERVAL 5 SECONDS PRECEDING
    -> HAVING cnt > 10;
            

window_clause ::= WINDOW window_name AS ( range_expr ), WINDOW ...
        

Windows may be defined with a window clause at the end of a SELECT statement. A window_name may be substituted for a range_expr anywhere in the SELECT statement:

rtsql> SELECT COUNT(*) FROM foo OVER mywin
    -> WINDOW mywin AS (RANGE INTERVAL 10 SECONDS PRECEDING);
          

The columns of a stream event are extracted from the event body. In Flume, events have additional properties (the originating host, priority, and timestamp). Each event may also be decorated with additional arbitrary named attributes.

Named attributes of an event can be accessed with the syntax "#attrname". This acts like a column of type BINARY.

For example, to select events with the interesting attribute:

rtsql> SELECT * FROM foo WHERE #interesting IS NOT NULL;
          

Event attributes are defined as a STRING key and a BINARY value. To use these values as strings, use the BIN2STR() function:

rtsql> SELECT * FROM foo WHERE BIN2STR(#x) = 'abc';
          

A set of functions allowing you to access the host, priority, and timestamp properties of each event are described in Section 9.1, “Event property accessor functions”.

For example, to select events only at the ERROR priority level:

rtsql> SELECT * FROM foo WHERE PRIORITY() = 'ERROR';
          

The priority field is also available as an integer. More urgent priorities have lower ordinal values ('FATAL' is 0). To select events at the WARN level and more urgent:

rtsql> SELECT * FROM foo WHERE PRIORITY_LEVEL() <= 2;
          

The BIGINT and INT types store integer values in 64- and 32-bit values, accordingly. The text representation of these values (e.g., in the delimited event format) is a base-10 integer.

The BOOLEAN type holds true/false values only. It cannot be coerced to or from an integer type. The text representation of these values are the UTF-8 strings true and false.

The FLOAT and DOUBLE types hold floating-point values. Operations on floating-point values may be imprecise, subject to the constraints of the IEEE floating-point format standard. All string values parsed by java.lang.Float.valueOf() and java.lang.Double.valueOf() may be used as the text representation of these values.

The PRECISE() type constructor defines a family of numeric types providing specific degrees of precision. The argument to the type constructor specifies the "scale" of the value; if positive, this is the number of significant digits to the right of the decimal place. If 0, all digits are significant. Operations on PRECISE values preserve the scale of the value.

The STRING type will hold a UTF-8 encoded character string of arbitrary length. (Events in Flume have a maximum length, which defaults to 32 KB. The contents of all columns together in an event is subject to this limit.)

The empty string is a legal value in rtsql. In delimited events, a NULL string is denoted by a configurable escape sequence which defaults to the two characters \N. Delimited events may not contain the delimiter character itself; the ability to escape these characters is future work.

The BINARY type holds a byte buffer of arbitrary bytes. When reading from an Avro input source, any byte array can be specified. When reading from text-based inputs, the byte array will be the bytes representing the UTF-8 encoding of the string input.

If coerced to the STRING type (either implicitly, or explicitly through the BIN2STR() function, the UTF-8 character set will be applied to the bytes.

The STR2BIN() function will do the reverse, returning a BINARY object that explicitly represents the UTF-8 bytes of its input string argument.

Descriptions of functions that manipulate binary data are available in Section 9.4, “Functions to work with binary data”.

rtsql employs a notion of a TIMESTAMP distinct from other numerical values. A TIMESTAMP contains two fields; a milliseconds part, and a nanoseconds part. The milliseconds part holds the number of milliseconds since the UNIX epoch (00:00:00 UTC, Jan 1, 1970). The nanoseconds part holds the number of nanoseconds after the specified millisecond and should not exceed 1,000,000.

Despite this internal precision, the delimited event format parses timestamps as a single 64-bit base-10 integer corresponding to the milliseconds part, and the nanoseconds part is 0. Furthermore, FlumeBase's internal notion of time and event-ordering works only at the granularity of milliseconds. The Avro event format allows the nanoseconds part to be specified. Even in this case, a column specified by timestamp.col (see Section 6.1.1.1, “Designated timestamp columns”) will associate only the milliseconds part of a timestamp with the event itself.

The LIST type allows you to represent a list of values as the value for a single column. The values in a list may be interacted with as a group, or individual values may be extracted from the list.

Values in a list must all have the same type. This type is specified with the syntax: LIST<t>, where t is the specification of another type. For example: LIST<INT>, or LIST<STRING NOT NULL>. You may not specify LIST without specifying the types of the members of the list.

Any type may appear in the parameter to the LIST type constructor. It is legal to specify, for example, LIST<LIST<INT NOT NULL>>. Note that LIST<INT> NOT NULL is different from LIST<INT NOT NULL> and LIST<INT NOT NULL> NOT NULL -- though all three of these are legal.

Lists can be parsed from events with the delimited event format; their elements are separated by pipe characters ("|"). You can override this by specifying the list.delim property of the event format.

Several functions exist to construct and manipulate lists; a reference is provided in Section 9.3, “Functions that operate on lists”.

Values of some types may be coerced to act as another type. This is performed automatically as necessary; no explicit type-casting operators are used (nor are they provided by the language). Coercion may only occur when moving to a "broader" type; this is called promotion in rtsql. The value itself may be modified to conform to the specific domain of the target type, but this is only performed if the modification would not lose data.

For example, an expression adding an INT value with a BIGINT value will provide a BIGINT result. The INT argument will be promoted to BIGINT and then added to the other BIGINT. Similarly, concatenating a STRING and an INT will coerce the integer into a string representation, and then concatenate it with the string.

Numeric values may be promoted to any broader numeric type. From narrowest to broadest, the numeric types are INT, BIGINT, FLOAT, and DOUBLE. PRECISE types may be promoted to any broader PRECISE type. INT and BIGINT promote to PRECISE(0), FLOAT promotes to PRECISE(24), and DOUBLE promotes to PRECISE(53).

All scalar types may be promoted to STRING. The result of coercing a value to STRING is the string representation of the value, as defined in the previous subsections.

Any type X NOT NULL may be promoted to its nullable counterpart.

A type LIST<X> may promote to LIST<Y> if X promotes to Y.

Some functions may operate over a variety of types. For example, the sum function will add values of any numeric type. Since DOUBLE is the widest numeric type available, the type of the argument column for sum could be marked as DOUBLE. But that would promote every input to sum to a DOUBLE value, and the output would also be DOUBLE. This is not the case; using the DESCRIBE statement, we can see the type of the sum function is:

rtsql> DESCRIBE sum;
sum ((var('a, constraints={TYPECLASS_NUMERIC})) -> var('a, constraints={TYPECLASS_NUMERIC}))
          

In order to allow the sum function to accept a variety of input types, its input type is specified as a type variable with a universal type; this is denoted by the var('a) type. The name of this type variable is 'a and pronounced "alpha." Type variables are bound to concrete types (e.g., INT) when the rtsql statement is compiled. The same type variable specifies the argument and return types of the sum; this way, it will return the same type as it receives as input. (This can be verified by observing that 'a is also the name of the output type.)

The sum function cannot sum a set of strings, however. Nor can it sum boolean values. Type variables may be specified with a set of constraints; the type it is bound to must conform to these constraints. The sum function can operate only over numeric values. To allow this constraint specification, rtsql provides a set of type classes which are sets of types. Type classes are not themselves concrete types; a column cannot be specified with a type class. The set of type classes are listed in Table 5, “Type classes and constraints in rtsql”.

Some functions allow a variable-length argument list. They may take a number of required arguments, and may then have a list of arguments of arbitrary length. The to_list() function, for example, will construct a list out of all its arguments. The following selects an empty list:

rtsql> select to_list() from x;
timestamp      to_list()
1306906081936  []
          

Whereas the following will construct a list of 3 elements:

rtsql> select to_list(42,211,312) from x;
timestamp        to_list(42, 211, 312)
1306906138072    [42, 211, 312]
          

Variable-length argument arrays are denoted by an ellipse (...) in the type signature of the function. For example:

rtsql> describe to_list;
to_list ((var('a, constraints={TYPECLASS_ANY})...) -> LIST<var('a, constraints={TYPECLASS_ANY})> NOT NULL)