[BUG] Found duplicate data when using the iceberge about starrocks HOT 7 CLOSED

same here #41267 (comment)

from starrocks.

hi, what's your starrocks version?

from starrocks.

same here #41267 (comment)

same here #41267 (comment)

Thanks! I will go study

from starrocks.

hi, what's your starrocks version?

The starrocks version is 3.1.10

from starrocks.

@mlbzssk hi, how do you upsert your table. could you share the reproduce process? I can't reproduce the issue using 3.1.10

from starrocks.

@mlbzssk hi, how do you upsert your table. could you share the reproduce process? I can't reproduce the issue using 3.1.10
@gaydba
@stephen-shelby
I also use the 3.1.10。
As i can see， this problem similar to apache/iceberg#6153
And the version which I use have the patch #41267 (comment).
the reproduce procedure is like:
1、create a mysql table，insert some data
create table mysql_table(id int primary key ,addr varchar(50),ts varchar(50) );
insert into mysql_table values(1,"aaa","2024-04-30");
insert into mysql_table values(2,"bb","2024-04-30");
insert into mysql_table values(3,"ccc","2024-04-30");
2、use spark-sql create a table with not primary key
CREATE TABLE spark_table (
id int,
addr string,
ts string) USING iceberg PARTITIONED BY (ts)
TBLPROPERTIES('format-version'='2', 'write.upsert.enabled'='true');
3、use flink sql create 2 tables
CREATE TABLE default_catalog.default_database.flink_table_to_mysql (
id int,
addr string,
ts string,
PRIMARY KEY(id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = '',
'port' = '',
'username' = '',
'password' = '',
'database-name' = 'iceberg_query_data',
'table-name' = 'mysql_table'
);

CREATE TABLE flink_table_to_spark (
id INT COMMENT 'unique id',
addr STRING NOT NULL,
ts string,
PRIMARY KEY(id,ts) NOT ENFORCED
) PARTITIONED BY (ts)
WITH (
'connector'='iceberg',
'catalog-name'='hive_prod',
'catalog-database'='default',
'catalog-table'='spark_table',
'uri'='thrift://',
'warehouse'='hdfs://'
);
4、let the data flow from default_catalog.default_database.flink_table_to_mysql to flink_table_to_spark
INSERT INTO
flink_table_to_spark /*+ OPTIONS('upsert-enabled'='true') */
SELECT * from default_catalog.default_database.flink_table_to_mysql;
5、update one row data in mysql
update mysql_table set addr='hahahaha' where id = '1';
6、query the data with starrocks and spark，you will find the result of the starrocks have 2 rows，but the spark only have 1 row.
CREATE EXTERNAL CATALOG hive_catalog_iceberg
PROPERTIES
(
"type" = "iceberg",
"iceberg.catalog.type" = "hive",
"hive.metastore.uris" = "thrift://"
);

from starrocks.

resolved

from starrocks.