Flink SQL Query

Description

The Flink SQL Query node (flink-sql-query) runs a Flink SQL query over the records flowing into the node. It lets you use the parts of Flink SQL that have no dedicated Nussknacker component: window table-valued functions, OVER aggregations, Window Top-N, pattern recognition with MATCH_RECOGNIZE, or SQL joins of several branches.

• The variables available before the node (e.g. #input and the variables defined upstream) are the columns of an input table. A record variable becomes a column of the ROW type, so its fields are accessed with a dot: record.input.amount.
• Each row returned by the query is emitted as a new record. The row is available in the output variable as a record whose fields are the columns returned by the query.
• The node clears all upstream variables - after the node, #input and the other variables are no longer available. Anything needed downstream has to be returned by the SELECT.

The node is available only in the Streaming processing mode, for unbounded streams.

Parameters and configuration

Name	Description
Query	The Flink SQL query producing the node's output. Default: SELECT r.* FROM record r. With one incoming branch the upstream variables are the columns of the record table; with several, each branch is a table named by its Branch alias. The query is fixed at design time - it cannot be built from the values of the processed records.
Branch alias	Shown only when the node has more than one incoming branch, one value per branch. The name under which the branch is available as a table in the query. It must be unique across branches, start with a letter or underscore, and contain only letters, digits and underscores.
Output variable name	The name of the variable holding the row returned by the query.

Input tables

Table	When	Columns
record	the node has one incoming branch	one column per variable available before the node, plus record_time
<Branch alias>	the node has several incoming branches (it acts as a join node)	the variables of that branch, plus record_time

record_time (type TIMESTAMP_LTZ(3)) holds the record's timestamp (event time) and carries the watermark of the source, so it can be used as the time attribute in windows, OVER ... ORDER BY record_time or MATCH_RECOGNIZE. Because of that, a variable named record_time must not be defined before the node - the node reports an error.

What SQL is allowed

The query must be a single query statement: SELECT (also with WITH), UNION, INTERSECT, EXCEPT, VALUES or ORDER BY. DDL and other statements (CREATE TABLE, INSERT INTO, SHOW FUNCTIONS, ...) are rejected. The query is validated by the Flink planner when the scenario is validated, so unknown columns, unknown functions or type mismatches are reported as errors on the Query field, with the planner's message. Built-in Flink SQL functions can be used. Flink user-defined functions configured in Nussknacker are not registered in the query - they are available as separate nodes.

Change Data Capture (CDC) and the #cdcOperation variable

A query can produce an updating result (a changelog) - e.g. a global aggregation without a window (SELECT COUNT(*) AS cnt FROM record), a GROUP BY without a window, or an outer join. In that case each emitted record represents a change of the result, and the node additionally exposes the #cdcOperation variable (type CdcOperation) with one of INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE. An update of a result row is emitted as two records: UPDATE_BEFORE with the previous version and UPDATE_AFTER with the new one. When only the latest values are needed downstream, drop the UPDATE_BEFORE records with a Filter on #cdcOperation.

When the query result is append-only (e.g. projections, filters, window aggregations or MATCH_RECOGNIZE over append-only input), #cdcOperation is not defined after the node.

The node also accepts changelog input: when the records entering the node carry a #cdcOperation variable of the CdcOperation type - e.g. records read by the Table source from a CDC table - the input table is treated as a changelog, and cdcOperation is not a column of the input table. A cdcOperation variable of any other type (e.g. a String) is an ordinary column.

Examples

Count events per key in one-day tumbling windows:

SELECT
  input.customerId AS customerId,
  window_start,
  COUNT(*) AS eventCount
FROM TABLE(
  TUMBLE(TABLE record, DESCRIPTOR(record_time), INTERVAL '1' DAY)
)
GROUP BY input.customerId, window_start, window_end

The output variable (e.g. #out) is then a record with the fields customerId, window_start and eventCount.

Running sum with an OVER aggregation:

SELECT
  input.amount AS amount,
  SUM(input.amount) OVER (ORDER BY record_time) AS runningSum
FROM record

Detect three consecutive increasing values with MATCH_RECOGNIZE:

SELECT first_val, second_val, third_val
FROM (
  SELECT input.num AS num, record_time
  FROM record
)
MATCH_RECOGNIZE (
  ORDER BY record_time
  MEASURES
    A.num AS first_val,
    B.num AS second_val,
    C.num AS third_val
  ONE ROW PER MATCH
  AFTER MATCH SKIP PAST LAST ROW
  PATTERN (A B C)
  DEFINE
    A AS TRUE,
    B AS B.num > A.num,
    C AS C.num > B.num
)

Join three branches with the aliases orders, products and categories:

SELECT o.input.orderId AS orderId, p.input.name AS product, c.input.name AS category
FROM orders o
JOIN products p ON o.input.productId = p.input.id
JOIN categories c ON p.input.categoryId = c.input.id

Additional considerations

• The node is stateful: windows, aggregations and joins keep their state in Flink state, and this state is restored from a savepoint when the scenario is redeployed. Changing the query may make the state incompatible.
• A regular (non-windowed) join or aggregation keeps state for every key indefinitely; prefer window aggregations and interval/window joins for unbounded streams.
• The node can't be mocked in scenario tests - it has to be fed real records.