Joining continuous queries in Flink SQL

Question

I'm trying to join two continuous queries, but keep running into the following error:

Rowtime attributes must not be in the input rows of a regular join. As a workaround you can cast the time attributes of input tables to TIMESTAMP before.\nPlease check the documentation for the set of currently supported SQL features.

Here's the table definition:

CREATE TABLE `Combined` (
    `machineID` STRING,
    `cycleID` BIGINT,
    `start` TIMESTAMP(3),
    `end` TIMESTAMP(3),
    WATERMARK FOR `end` AS `end` - INTERVAL '5' SECOND,
    `sensor1` FLOAT,
    `sensor2` FLOAT
)

and the insert query

INSERT INTO `Combined` 
SELECT
    a.`MachineID`,
    a.`cycleID`,
    MAX(a.`start`) `start`,
    MAX(a.`end`) `end`,
    MAX(a.`sensor1`) `sensor1`,
    MAX(m.`sensor2`) `sensor2`
FROM `Aggregated` a, `MachineStatus` m
WHERE 
    a.`MachineID` = m.`MachineID` AND 
    a.`cycleID` = m.`cycleID` AND 
    a.`start` = m.`timestamp`
GROUP BY a.`MachineID`, a.`cycleID`, SESSION(a.`start`, INTERVAL '1' SECOND)

In the source tables Aggregated and MachineStatus, the start and timestamp columns are time attributes with a watermark.

I've tried casting the input rows of the join to timestamps, but that didn't fix the issue and would mean that I cannot use SESSION, which is supposed to ensure that only one data point gets recorded per cycle.

Any help is greatly appreciated!

Daniel Lenz Daniel Lenz · Accepted Answer · 2020-08-24T06:34:45

I investigated this a little further and noticed that the GROUP BY statement doesn't make sense in that context.

Furthermore, the SESSION can be replaced by a time window, which is the more idiomatic approach.

INSERT INTO `Combined` 
SELECT
    a.`MachineID`,
    a.`cycleID`,
    a.`start`,
    a.`end`,
    a.`sensor1`,
    m.`sensor2`
FROM `Aggregated` a, `MachineStatus` m
WHERE 
    a.`MachineID` = m.`MachineID` AND 
    a.`cycleID` = m.`cycleID` AND 
    m.`timestamp` BETWEEN a.`start` AND a.`start` + INTERVAL '0' SECOND

To understand the different ways to join dynamic tables, I found the Ververica SQL training extremely helpful.

Joining continuous queries in Flink SQL

1 Answers