You are viewing a plain text version of this content. The canonical link for it is here.

Posted to user@flink.apache.org by elakiya udhayanan <la...@gmail.com> on 2023/10/19 07:04:20 UTC

Flink SQL exception on using cte

Hi Team,

I have a Flink job which uses the upsert-kafka connector to consume the
events from two different Kafka topics (confluent avro serialized) and
write them to two different tables (in Flink's memory using the Flink's SQL
DDL statements).

I want to correlate them using the SQL join statements and for this I am
trying to use the cte expressions like below (sample): But getting
exception as below

 *org.apache.flink.table.api.SqlParserException: SQL parse failed.
Incorrect syntax near the keyword 'INSERT'*

WITH p1 AS ( SELECT empId FROM employee )
 FROM p1
INSERT INTO correlate
SELECT pod;

Please let me know if queries with cte are supported in Apache Flink.

Thanks,
Elakiya

RE: Flink SQL exception on using cte

Posted by Aniket Sule <an...@netwitness.com>.

Hello,
I have been able to use queries with cte in this syntax –
INSERT INTO t1
      WITH cte1 AS (SELECT
                   ....),
                                 cte2 AS (SELECT
                   ....)
                               (SELECT
                  *
              FROM
                  cte1 AS a, cte2 as b .....
              );

Hope this helps you.

Regards,
Aniket Sule

From: elakiya udhayanan <la...@gmail.com>
Sent: Thursday, October 19, 2023 3:04 AM
To: user@flink.apache.org
Subject: Flink SQL exception on using cte

CAUTION:External email. Do not click or open attachments unless you know and trust the sender.

Hi Team,

I have a Flink job which uses the upsert-kafka connector to consume the events from two different Kafka topics (confluent avro serialized) and write them to two different tables (in Flink's memory using the Flink's SQL DDL statements).

I want to correlate them using the SQL join statements and for this I am trying to use the cte expressions like below (sample): But getting  exception as below

 org.apache.flink.table.api.SqlParserException: SQL parse failed. Incorrect syntax near the keyword 'INSERT'


WITH p1 AS ( SELECT empId FROM employee )

 FROM p1

INSERT INTO correlate

SELECT pod;
Please let me know if queries with cte are supported in Apache Flink.

Thanks,
Elakiya
Caution: External email. Do not click or open attachments unless you know and trust the sender.

Re: Flink SQL exception on using cte

Posted by elakiya udhayanan <la...@gmail.com>.

Thanks Robin and Aniket for the suggestions you have given.

Will try and update on the same.

Thanks,
Elakiya

On Fri, Oct 20, 2023 at 2:34 AM Robin Moffatt <ro...@decodable.co> wrote:

> CTEs are supported, you can see an example in the docs [1] [2]. In the
> latter doc, it also says
>
> > CTEs are supported in Views, CTAS and INSERT statement
>
> So I'm just guessing here, but your SQL doesn't look right.
> The CTE needs to return a column called `pod`, and the `FROM` clause for
> the `SELECT` should be after it, not before the `INSERT`.
>
> i.e. something like this instead:
>
> WITH p1 AS ( SELECT empId AS pod FROM employee )
> INSERT INTO correlate
> SELECT pod FROM p1;
>
>
> Hope that helps,
>
> Robin
>
> [1]:
> https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/sql/queries/with/
> [2]:
> https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/hive-compatibility/hive-dialect/queries/cte/
>
> On Thu, 19 Oct 2023 at 08:05, elakiya udhayanan <la...@gmail.com>
> wrote:
>
>> Hi Team,
>>
>> I have a Flink job which uses the upsert-kafka connector to consume the
>> events from two different Kafka topics (confluent avro serialized) and
>> write them to two different tables (in Flink's memory using the Flink's SQL
>> DDL statements).
>>
>> I want to correlate them using the SQL join statements and for this I am
>> trying to use the cte expressions like below (sample): But getting
>> exception as below
>>
>>  *org.apache.flink.table.api.SqlParserException: SQL parse failed.
>> Incorrect syntax near the keyword 'INSERT'*
>>
>> WITH p1 AS ( SELECT empId FROM employee )
>>  FROM p1
>> INSERT INTO correlate
>> SELECT pod;
>>
>> Please let me know if queries with cte are supported in Apache Flink.
>>
>> Thanks,
>> Elakiya
>>
>

Re: Flink SQL exception on using cte

Posted by Robin Moffatt <ro...@decodable.co>.

CTEs are supported, you can see an example in the docs [1] [2]. In the
latter doc, it also says

> CTEs are supported in Views, CTAS and INSERT statement

So I'm just guessing here, but your SQL doesn't look right.
The CTE needs to return a column called `pod`, and the `FROM` clause for
the `SELECT` should be after it, not before the `INSERT`.

i.e. something like this instead:

WITH p1 AS ( SELECT empId AS pod FROM employee )
INSERT INTO correlate
SELECT pod FROM p1;


Hope that helps,

Robin

[1]:
https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/sql/queries/with/
[2]:
https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/hive-compatibility/hive-dialect/queries/cte/

On Thu, 19 Oct 2023 at 08:05, elakiya udhayanan <la...@gmail.com> wrote:

> Hi Team,
>
> I have a Flink job which uses the upsert-kafka connector to consume the
> events from two different Kafka topics (confluent avro serialized) and
> write them to two different tables (in Flink's memory using the Flink's SQL
> DDL statements).
>
> I want to correlate them using the SQL join statements and for this I am
> trying to use the cte expressions like below (sample): But getting
> exception as below
>
>  *org.apache.flink.table.api.SqlParserException: SQL parse failed.
> Incorrect syntax near the keyword 'INSERT'*
>
> WITH p1 AS ( SELECT empId FROM employee )
>  FROM p1
> INSERT INTO correlate
> SELECT pod;
>
> Please let me know if queries with cte are supported in Apache Flink.
>
> Thanks,
> Elakiya
>