[FLINK-22853] Flink SQL functions max/min/sum return duplicated records

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Not A Problem
• Affects Version/s: 1.12.1
• Component/s: Connectors / JDBC, Table SQL / API
• Assignee: Unassigned
• Reporter: Raypon Wang
• Created: 02/Jun/21 10:53
• Updated: 07/Jun/21 10:38
• Resolved: 07/Jun/21 10:38

Description

mysql data：

id    offset

1      1

1      3

1      2

flinksql code：(I used flink-connector-jdbc_2.12:1.12.1)

 

 

object FlinkSqlOnJdbcForMysql {
def main(args: Array[String]): Unit =

{ val settings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build() val tableEnvironment = TableEnvironment.create(settings) tableEnvironment.executeSql("" + "CREATE TABLE test (" + " id BIGINT," + " `offset` INT," + " PRIMARY KEY (id) NOT ENFORCED" + ") WITH (" + " 'connector' = 'jdbc'," + " 'driver' = 'com.mysql.cj.jdbc.Driver'," + " 'url' = 'jdbc:mysql://127.0.0.1:3306/test?&serverTimezone=Asia/Shanghai'," + " 'username' = 'root'," + " 'password' = 'Project.03'," + " 'table-name' = 'test'," + " 'scan.fetch-size' = '1000'," + " 'scan.auto-commit' = 'true'" + ")") tableEnvironment.executeSql( "select id,max(`offset`) from test group by id").print() }

}

 

result:

---------------------------------+

id

EXPR$1

---------------------------------+

1	1
1	3
1	2

---------------------------------+

Result of max/min/sum is duplicated,

but avg/count/last_value/first_value is ok.