NebulaStream Query API Guide

NebulaStream provides stream processing through a declarative, SQL-like query language. This guide explains core concepts and how to create queries.

The fundamental data flow is simple:

1. Sources ingest data into the system.
2. Operators process and transform data in-flight.
3. Sinks emit results to external systems (databases, files) or to the console.

Let’s start with key terminology.

Core Concepts Glossary

A Complete Query Example

Queries can be submitted as YAML specifications or SQL statements. Below is a complete example from the Linear Road Benchmark, which we’ll break down section by section.

query: |
  SELECT start, end, highway, direction, positionDiv5280, AVG(speed) AS avgSpeed
  FROM (SELECT creationTS, highway, direction, position / 5280 AS positionDiv5280, speed FROM lrb)
  GROUP BY (highway, direction, positionDiv5280)
  WINDOW SLIDING(creationTS, SIZE 5 MINUTES, ADVANCE BY 1 SEC)
  HAVING avgSpeed < 40
  INTO csv_sink;

sinks:
  - name: csv_sink
    host: localhost:8080
    schema:
      - name: start
        type: UINT64
      - name: end
        type: UINT64
      - name: highway
        type: INT16
      - name: direction
        type: INT16
      - name: positionDiv5280
        type: INT32
      - name: avgSpeed
        type: FLOAT64
    type: File
    config:
      output_format: CSV
      file_path: "<path>"
      append: false
    parser_config:
      quote_strings: false

logical:
  - name: lrb
    schema:
      - name: creationTS
        type: UINT64
      - name: vehicle
        type: INT16
      - name: speed
        type: FLOAT32
      - name: highway
        type: INT16
      - name: lane
        type: INT16
      - name: direction
        type: INT16
      - name: position
        type: INT32

physical:
  - logical: lrb
    type: TCP
    host: localhost:8080
    parser_config:
      type: CSV
      tuple_delimiter: "\n"
      field_delimiter: ","
    source_config:
      socket_host: localhost
      socket_port: 50501
      socket_buffer_size: 65536
      flush_interval_ms: 100
      connect_timeout_seconds: 60
  - logical: lrb
    host: localhost:8080
    type: File
    parser_config:
      type: JSON
    source_config:
      file_path: lrb.json
workers:
  - host: localhost:8080
    data_address: localhost:9090

This YAML can be sent to nes-cli to register or run the query.

Here’s the equivalent SQL syntax:

CREATE WORKER 'localhost:8080' SET ('localhost:9090' AS DATA);
CREATE LOGICAL SOURCE lrb(
  creationTS UINT64,
  vehicle INT16,
  speed FLOAT32,
  highway INT16,
  lane INT16,
  direction INT16,
  position INT32
);

CREATE PHYSICAL SOURCE FOR lrb TYPE TCP SET(
  'localhost:8080' AS "SOURCE"."HOST",
  'localhost' as "SOURCE".SOCKET_HOST,
  50501 as "SOURCE".SOCKET_PORT,
  65536 as "SOURCE".SOCKET_BUFFER_SIZE,
  100 as "SOURCE".FLUSH_INTERVAL_MS,
  60 as "SOURCE".CONNECT_TIMEOUT_SECONDS,
  'CSV' as INPUT_FORMATTER."TYPE",
  '\n' as INPUT_FORMATTER.TUPLE_DELIMITER,
  ',' as INPUT_FORMATTER.FIELD_DELIMITER
);

CREATE PHYSICAL SOURCE FOR lrb TYPE File SET(
  'localhost:8080' AS "SOURCE"."HOST",
  'lrb.json' as "SOURCE".FILE_PATH,
  'JSON' as INPUT_FORMATTER."TYPE"
);

CREATE SINK csv_sink(
  start UINT64 NOT NULL,
  end UINT64 NOT NULL,
  highway INT16,
  direction INT16,
  positionDiv5280 INT32,
  avgSpeed FLOAT64
) TYPE File SET(
  'localhost:8080' AS "SINK"."HOST",
  '<path>' as "SINK".FILE_PATH,
  'CSV' as "SINK".OUTPUT_FORMAT,
  FALSE as "SINK".APPEND,
  FALSE as "OUTPUT_FORMATTER".QUOTE_STRINGS
);

SELECT start, end, highway, direction, positionDiv5280, AVG(speed) AS avgSpeed
FROM (SELECT creationTS, highway, direction, position / 5280 AS positionDiv5280, speed FROM lrb)
GROUP BY (highway, direction, positionDiv5280)
WINDOW SLIDING(creationTS, SIZE 5 MINUTES, ADVANCE BY 1 SEC)
HAVING avgSpeed < 40
INTO csv_sink;

Anatomy of a Query

Data Sources: Logical and Physical

NebulaStream separates logical and physical sources to provide flexible data ingestion.

Logical Sources

A logical source is like a table definition in a traditional database. It provides an abstract description of a data stream with a name and schema.

• The name references the stream in your query’s FROM clause.
• The schema defines the structure of data records (tuples), listing each field’s name and data type.

Operators automatically infer their schemas from the source, so you only define it once. Incoming data must strictly match this schema, including field order. Any mismatch terminates the query.

Here is the logical source definition from our example:

CREATE LOGICAL SOURCE lrb(
  creationTS UINT64,
  vehicle INT16,
  speed FLOAT32,
  highway INT16,
  lane INT16,
  direction INT16,
  position INT32
);

This defines the lrb source used in the query’s FROM clause. The schema specifies seven fields per record. Multiple logical sources can be defined and combined with JOIN or UNION.

Physical Sources

A physical source specifies how and where to ingest data for a logical source. Each logical source can have multiple physical sources, allowing a single stream to aggregate data from heterogeneous endpoints.

Supported physical source types:

• File
• TCP
• MQTT — Source MQTT documentation coming soon.
• Generator — Source Generator documentation coming soon.

In our example, we define two physical sources that both feed the lrb logical source:

CREATE PHYSICAL SOURCE FOR lrb TYPE TCP SET(
  'localhost:8080' AS "SOURCE"."HOST",
  'localhost' as "SOURCE".SOCKET_HOST,
  50501 as "SOURCE".SOCKET_PORT,
  65536 as "SOURCE".SOCKET_BUFFER_SIZE,
  100 as "SOURCE".FLUSH_INTERVAL_MS,
  60 as "SOURCE".CONNECT_TIMEOUT_SECONDS,
  'CSV' as INPUT_FORMATTER."TYPE",
  '\n' as INPUT_FORMATTER.TUPLE_DELIMITER,
  ',' as INPUT_FORMATTER.FIELD_DELIMITER
);

CREATE PHYSICAL SOURCE FOR lrb TYPE File SET(
  'localhost:8080' AS "SOURCE"."HOST",
  'lrb.json' as "SOURCE".FILE_PATH,
  'JSON' as INPUT_FORMATTER."TYPE"
);

As you can see, one source reads CSV-formatted data from a TCP socket, while the other reads JSON-formatted data from a file. Both produce tuples that conform to the lrb schema.

The CSV file might look like this:

creationTS,vehicle,speed,highway,lane,direction,position
1234567890,101,65.5,1,2,0,15840
1234567891,102,70.2,1,3,0,21120
1234567892,103,55.8,2,1,1,10560
1234567893,101,68.3,1,2,0,16896

Each physical source requires configuration for:

• The specific connector (e.g., file path or TCP socket details) via SOURCE.* parameters.
• The data’s input format (e.g., CSV or JSON) and delimiters via INPUT_FORMATTER.* parameters.

The query itself remains completely decoupled from these physical details. You can add, remove, or change physical sources without touching the query logic.

Data Sinks: Defining the Output

Sinks represent the destination for query results. Currently, a query must have exactly one sink.

CREATE SINK csv_sink(
  start UINT64 NOT NULL,
  end UINT64 NOT NULL,
  highway INT16,
  direction INT16,
  positionDiv5280 INT32,
  avgSpeed FLOAT64
) TYPE File SET(
  'localhost:8080' AS "SINK"."HOST",
  '<path>' as "SINK".FILE_PATH,
  'CSV' as "SINK".OUTPUT_FORMAT,
  FALSE as "SINK".APPEND
);

The sink name (csv_sink) must match the name used in the query’s INTO clause.

Available sink types include:

• File: Writes results to a file, either overwriting or appending.
• Print: Writes results to standard output (stdout).
• Void — Sink Void documentation coming soon.
• MQTT — Sink MQTT documentation coming soon.

The SET clause specifies the output details. For a File sink, this includes the file path and the data format for the output.

The HOST configuration parameter specifies the worker node, identified by its gRPC address, which hosts the physical source/sink.

• The sink itself can be configured via SINK.* parameters.
• The output formatter can be configured via OUTPUT_FORMATTER.* parameters.

Input Formatters

Tuples can arrive in a variety of formats. We distinguish two broad categories:

• Text-based formats (JSON, CSV, XML, YAML, etc.)
• Binary formats (Avro, Parquet, Protobuf, etc.)

Input formatters convert byte streams from source connectors into the native in-memory representation used by query-compiled operators. The format is specified via INPUT_FORMATTER.* parameters in each physical source:

CREATE PHYSICAL SOURCE FOR source_name TYPE TCP SET(
  'CSV' as INPUT_FORMATTER."TYPE",
  '\n' as INPUT_FORMATTER.TUPLE_DELIMITER,
  ',' as INPUT_FORMATTER.FIELD_DELIMITER,
  ...
);

Currently, NebulaStream supports CSV and JSON input formats.

Output Formatters

The output formatter component converts records with values in our native in-memory format into the desired output format of a sink. They are employed by sinks that utilize the OUTPUT_FORMAT parameter to configure the format of the result tuples.

Out-of-the-box available output formats are:

• CSV
• JSON

Some output formats may be configurable via parameters. For instance, the bool parameter QUOTE_STRINGS controls how the CSVOutputFormatter represents strings. All required parameters can be specified via OUTPUT_FORMATTER.* in each sink.

CREATE SINK sink_name TYPE FILE SET(
       'CSV' as "SINK".OUTPUT_FORMAT,
       TRUE as "OUTPUT_FORMATTER".QUOTE_STRINGS,
       ...
);

Identifiers and Quotation Marks

Identifiers are the names of SQL objects, including sources, sinks, fields, aliases, and configuration keys. Simple identifiers can be written without quotes. Double quotes allow identifiers to contain spaces or preserve their exact spelling:

SELECT "event type" AS "Event Type" FROM "input stream" INTO sink;

Double-quoted text refers to an identifier, while single-quoted text represents a string value:

SELECT "status" AS field_value, 'status' AS literal_value FROM stream INTO sink;

Here, "status" refers to a field named status, whereas 'status' is the literal string status.

Data Types

In NebulaStream, each field is associated with exactly one data type. This data type specifies the physical memory layout and valid operations on the field.

Supported data types:

• INT8
• UINT8
• INT16
• UINT16
• INT32
• UINT32
• INT64
• UINT64
• FLOAT32
• FLOAT64
• CHAR
• BOOLEAN
• VARSIZED

These types match primitive C++ data types. The numeric suffix denotes the bit width. VARSIZED supports arbitrary-length data like strings. For output types of arithmetical operations, we stick to the C++ standard, c.f.Integer Promotions and Conversion Ranks.

Numeric literals

Numeric constants can be written directly in query expressions:

SELECT speed * 3.6 AS speed_m_sec FROM s INTO sink
SELECT * FROM s WHERE count >= 10 AND delta > -5 INTO sink

NebulaStream infers a raw numeric literal’s data type from its value:

• Integer literals without a leading minus sign use the smallest unsigned integer type that can represent the value: UINT8, UINT16, UINT32, or UINT64.
• Negative integer literals use the smallest signed integer type that can represent the value: INT8, INT16, INT32, or INT64.
• Floating-point literals, including fractional and exponent notation such as 0.1, 42.0, .5, and 1E3, use FLOAT64.

Use an explicit type constructor when a query depends on an exact literal type:

SELECT UINT64(1) AS id, FLOAT32(3.6) AS scale FROM s INTO sink

String and boolean literals

String and boolean constants can also be written directly in query expressions:

SELECT 'hello' AS message, TRUE AS enabled FROM s INTO sink
SELECT * FROM s WHERE status == 'completed' AND enabled == false INTO sink

Raw string literals infer VARSIZED, including numeric-looking strings such as '123'. Raw TRUE and FALSE literals infer BOOLEAN; lowercase true and false are also supported. Explicit constructors remain available when desired:

SELECT VARSIZED('hello') AS message, BOOLEAN(TRUE) AS enabled FROM s INTO sink

Operators

An operator consumes an input stream and produces an output stream. We differentiate between stateless and stateful operators. Stateless operators produce output tuples without buffering the stream.

Operators are either unary (one input stream) or binary (two input streams). All operators produce a single output stream. Data flows from sources to a single sink via unary operators (selection, projection) or binary operators (join, union).

Stateless Operators

Projection

Projections are compositions of functions that are enumerated after the SELECT keyword.

SELECT a, b, c FROM s INTO sink

SELECT speed * 3.6 AS speed_m_sec FROM s INTO sink

SELECT CONCAT(firstName, lastName) AS firstNameLastName FROM nameStream INTO firstNameLastNameSink;

💡 Use an explicit type constructor when a constant needs an exact type.

SELECT FLOAT32(3.141) * r FROM stream INTO sink

Selection

Selections use the WHERE keyword to filter the input stream.

SELECT * FROM s WHERE t == 'sometext' INTO sink

Predicates can combine functions, comparisons, and SQL predicate syntax with AND, OR, and NOT:

SELECT * FROM s WHERE CEIL(speed) != 0 OR altitude == 0 INTO sink

SELECT * FROM transactions WHERE amount > 1000.0 AND status == 'completed' INTO sink

BETWEEN is inclusive and supports NOT BETWEEN:

SELECT * FROM transactions WHERE amount BETWEEN 100.0 AND 1000.0 INTO sink

SELECT * FROM transactions WHERE amount NOT BETWEEN 100.0 AND 1000.0 INTO sink

IN supports explicit, non-empty value lists and also supports NOT IN; IN subqueries are not supported:

SELECT * FROM s WHERE status IN ('queued', 'running') INTO sink

SELECT * FROM s WHERE status NOT IN ('failed', 'cancelled') INTO sink

IS NULL and IS NOT NULL test nullable values:

SELECT * FROM s WHERE optional_value IS NULL INTO sink

SELECT * FROM s WHERE optional_value IS NOT NULL INTO sink

IS NaN and IS NOT NaN test whether a numeric value is a floating-point not a number:

SELECT * FROM measurements WHERE reading IS NaN INTO sink

SELECT * FROM measurements WHERE reading IS NOT NaN INTO sink

The operand must be numeric. Applying IS NaN to a VARSIZED, BOOLEAN, or CHAR field is rejected when the query is registered. Integer operands are accepted but are never NaN, so IS NaN is always false for them. Infinities are not NaN.

💡 IS NaN follows SQL three-valued logic, see null handling for more details.

The equivalent function-call spelling ISNAN(x) is also available, which is what you need in a projection: SELECT ISNAN(reading) AS invalid FROM measurements INTO sink.

Union

Union combines two input streams with identical schema into one.

SELECT * FROM s UNION (SELECT * FROM t) INTO sink

SELECT user_id, action, timestamp FROM web_events 
UNION (SELECT user_id, action, timestamp FROM mobile_events) INTO sink

💡 Union does not deduplicate values as in classical relational algebra.

Stateful/Windowed Operators

Stateful operators require more context than a single tuple to produce an output. In batch systems, these operations would require all input data to be seen before emitting results. This is not feasible in stream processing systems that deal with unbounded datasets. Therefore, we chunk the stream up into windows.

Window Types

Two window types are supported:

Tumbling Windows

Tumbling windows chunk the stream into disjoint subsets, for example for timestamps (1...6) and a window size of 3 [1 2 3][4 5 6].

Syntax: WINDOW TUMBLING(<timestamp_field>, <size><unit>)

WINDOW TUMBLING(ts, SIZE 1 SEC) INTO sink

Sliding Windows

Sliding windows chunk the stream into overlapping subsets, for example: [1s 2s][2s 3s][3s 4s]

Syntax: WINDOW SLIDING(<timestamp_field>, SIZE <size><unit>, ADVANCE BY <size><unit>)

💡 The timestamp field needs to be of type UINT64, with a millisecond resolution.

WINDOW SLIDING(ts, SIZE 1 SEC, ADVANCE BY 100 MS) INTO sink

Window Measures

Two window measures are supported:

Event Time

Event time uses timestamps defined in the tuples themselves to assign them to the correct windows.

💡 For binary windowed operators like joins, the timestamp field must have the same name for both input streams.

Ingestion Time

Ingestion time assigns tuples to windows based on the timestamp when the tuple was first ingested into the system. We omit the timestamp field specifier in the window definition:

WINDOW TUMBLING(SIZE 1 MIN) INTO sink

Aggregation

Aggregations allow you to compute summary statistics over windows of data. Common aggregation functions include mathematical operations like MAX, MIN, SUM, AVG, and statistical functions like MEDIAN.

SELECT MAX(price) FROM bid GROUP BY ticker WINDOW SLIDING(ts, SIZE 10 SEC, ADVANCE BY 1 SEC) INTO sink

SELECT MEDIAN(oxygen_level) FROM health_sensor WINDOW TUMBLING(ts, SIZE 100 MS) INTO sink

SELECT COUNT(*) AS event_count, AVG(response_time) AS avg_response 
FROM api_requests 
GROUP BY endpoint 
WINDOW TUMBLING(ts, SIZE 5 MIN) INTO sink

Windowed aggregations support an optional GROUP BY clause to specify grouping keys. A HAVING clause applies filters to aggregated results.

SELECT ticker, MAX(price) AS max_price, MIN(price) AS min_price
FROM stock_quotes 
GROUP BY ticker 
WINDOW TUMBLING(ts, SIZE 1 MIN)
HAVING MAX(price) > 100.0 AND COUNT(*) >= 10 INTO sink

Join

Joins combine tuples from two input streams based on a condition within a window. Only tuples that satisfy the join predicate are included in the output.

SELECT * FROM s INNER JOIN (SELECT * FROM t) ON sid = tid WINDOW TUMBLING(ts, SIZE 1 MIN) INTO sink

SELECT order_id, customer_id, amount 
FROM orders
INNER JOIN (SELECT * FROM payments p) ON order_id = payments_order_id 
WINDOW SLIDING(ts, SIZE 30 SEC, ADVANCE BY 5 SEC) INTO sink

💡 Currently, the timestamp field is required to have the same name in both input streams.

Table-Valued Functions

MODEL_INFERENCE

MODEL_INFERENCE runs a registered ML model on each input tuple and appends the model’s output fields to the result. Models must be registered with CREATE MODEL before use.

Registering a model:

CREATE MODEL iris ('/path/to/iris.onnx')
INPUT (p1 FLOAT32, p2 FLOAT32, p3 FLOAT32, p4 FLOAT32)
OUTPUT (setosa FLOAT32, versicolor FLOAT32, virginica FLOAT32);

• The INPUT fields must match the model’s input tensor shape and types. Each field maps to one element of the input tensor.
• The OUTPUT fields must match the model’s output tensor shape and types.
• Only .onnx model files are supported. Models are compiled to IREE bytecode at first use.
• Only FLOAT32 tensor element types are currently supported.

Using a model in a query:

-- Direct stream input: each tuple's fields are fed to the model
SELECT * FROM MODEL_INFERENCE(iris, stream) INTO result;

The output schema contains all fields from the input stream followed by the model’s output fields.

Using a subquery as input:

-- Decode base64-encoded image data before feeding to a model
SELECT c0, c1, c2, c3, c4, c5, c6, c7, c8, c9
FROM MODEL_INFERENCE(mnist, (SELECT FROM_BASE64(pixels) AS pixels FROM stream))
INTO result;

When the model input is defined as VARSIZED, a single binary blob (e.g., raw tensor bytes) is passed directly to the model runtime. This is useful for image data or pre-packed tensors.

Nesting models (chaining inference):

-- Feed the output of one model into another
SELECT * FROM MODEL_INFERENCE(model_b, MODEL_INFERENCE(model_a, stream)) INTO result;

💡 MODEL_INFERENCE requires the IREE runtime library and the IREE compiler tools (iree-import-onnx, iree-compile) to be installed. If the tools are not available, model compilation will fail at query time.

Functions

A function (also known as scalar expression) specifies an operation on one or more fields. For example, SELECT a + b FROM stream INTO sink uses the ADD function. Every expression is a function, including field access and constants. We refer to input parameters as input functions.

Functions are either unary (one input) or binary (two inputs). ABS is a unary function, while + is a binary function.

Supported Functions

Base

Arithmetical

Boolean/Comparison

Other

TO_BASE64 and FROM_BASE64 convert between raw binary data (VARSIZED) and base64-encoded text (VARSIZED). They use OpenSSL’s EVP base64 implementation. These functions are particularly useful with MODEL_INFERENCE to pass binary tensor data through SQL queries (see MODEL_INFERENCE).

CAST converts a field from one type to another using standard SQL syntax: CAST(field AS TargetType). For numeric-to-numeric conversions (e.g., INT32 to FLOAT64), a direct type cast is performed. When casting from VARSIZED (string) to a numeric type, all characters except digits (0-9), ., +, -, e, and E (scientific notation) are stripped before parsing. For example, "1 234.56" becomes 1234.56 and "12?3" becomes 123.

SELECT CAST(price AS FLOAT64) FROM s INTO sink

SELECT CAST(text_value AS INT32) AS parsed_value FROM s INTO sink

We can combine functions into nested structures:

SELECT POW((x AS actual) - (y AS predicted), 2) FROM s INTO sink

This calculates the squared error between a prediction and ground truth. Query compilation traces expression trees at compile time, producing efficient machine code instead of runtime evaluation.

Aggregation