You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Guillermo <ko...@gmail.com> on 2025/03/14 11:18:55 UTC

TableAggregation, FlinkSql - Top2 example

I'm trying to execute the example to AggregationTable (
https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/table/functions/udfs/#table-aggregate-functions)
from FlinkSQL and I don't get it.

I have copied the example, compiled and created the function. When I try to
execute it:

Flink SQL> CREATE FUNCTION Top2 AS 'org.example.Top2';

Flink SQL> CREATE TABLE example_top (
eventTimestamp BIGINT ,
id INT,
data INT,
ts AS TO_TIMESTAMP_LTZ(eventTimestamp, 3),
WATERMARK FOR ts AS ts
) WITH (
'connector' = 'kafka',
..

Flink SQL> SELECT e.id, Top2(e.data1)
FROM example_top AS e
GROUP BY id;

[ERROR] Could not execute SQL statement. Reason:
java.lang.IllegalArgumentException: Type mismatch:
rel rowtype: RecordType(INTEGER msisdn, INTEGER EXPR$1) NOT NULL
equiv rowtype: RecordType(INTEGER msisdn,
*org.apache.flink.api.java.tuple.Tuple2(INTEGER, INTEGER)* EXPR$1) NOT NULL
Difference:
EXPR$1: INTEGER -> *org.apache.flink.api.java.tuple.Tuple2(INTEGER,
INTEGER)*

How should I use the function? I tried many ways to call it.

Re: TableAggregation, FlinkSql - Top2 example

Posted by Shengkai Fang <fs...@gmail.com>.
I think Flink SQL does't support to use table aggregate function and you
should use table api to use table aggregate function.

Best,
Shengkai

Guillermo <ko...@gmail.com> 于2025年3月14日周五 19:21写道：

> I'm trying to execute the example to AggregationTable (
> https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/table/functions/udfs/#table-aggregate-functions)
> from FlinkSQL and I don't get it.
>
> I have copied the example, compiled and created the function. When I try
> to execute it:
>
> Flink SQL> CREATE FUNCTION Top2 AS 'org.example.Top2';
>
> Flink SQL> CREATE TABLE example_top (
> eventTimestamp BIGINT ,
> id INT,
> data INT,
> ts AS TO_TIMESTAMP_LTZ(eventTimestamp, 3),
> WATERMARK FOR ts AS ts
> ) WITH (
> 'connector' = 'kafka',
> ..
>
> Flink SQL> SELECT e.id, Top2(e.data1)
> FROM example_top AS e
> GROUP BY id;
>
> [ERROR] Could not execute SQL statement. Reason:
> java.lang.IllegalArgumentException: Type mismatch:
> rel rowtype: RecordType(INTEGER msisdn, INTEGER EXPR$1) NOT NULL
> equiv rowtype: RecordType(INTEGER msisdn,
> *org.apache.flink.api.java.tuple.Tuple2(INTEGER, INTEGER)* EXPR$1) NOT NULL
> Difference:
> EXPR$1: INTEGER -> *org.apache.flink.api.java.tuple.Tuple2(INTEGER,
> INTEGER)*
>
> How should I use the function? I tried many ways to call it.
>