waywtdcc created HUDI-6217:
------------------------------
Summary: Spark reads the deleted data whose _hoodie_operation is D
Key: HUDI-6217
URL: https://issues.apache.org/jira/browse/HUDI-6217
Project: Apache Hudi
Issue Type: Bug
Components: flink-sql, spark-sql
Affects Versions: 0.12.3, 0.13.0
Reporter: waywtdcc
Attachments: image-2023-05-16-11-34-54-045.png
Spark reads the deleted data whose _hoodie_operation is D。
flink:
{code:java}
CREATE TABLE cdc_test (
id INT,
name STRING,
age int,
PRIMARY KEY(id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = '10.252.226.4',
'port' = '33006',
'username' = 'root',
'password' = 'xxx',
'database-name' = 'data_collection',
'table-name' = 'cdc_test');
CREATE CATALOG myhudi WITH(
'type' = 'hudi',
'default-database' = 'test_ods',
'mode' = 'hms',
'hive.conf.dir' = 'hdfs:///flink/flink-conf'
);
drop table if exists myhudi.test_hudi3.hudi_datagen_incre_mysql_cdc;
create table if not exists myhudi.test_hudi3.hudi_datagen_incre_mysql_cdc
( id INT,
name string,
ts3 timestamp(3)
,PRIMARY KEY (`id`) NOT ENFORCED
)
with(
'connector' = 'hudi',
'table.type' = 'MERGE_ON_READ',
'compaction.tasks' = '1',
'write.tasks' = '1',
'index.bootstrap.enabled' = 'true',
'hive_sync.skip_ro_suffix' = 'true',
'compaction.delta_commits' = '100',
'compaction.async.enabled' = 'true',
'changelog.enabled' = 'true',
'index.type' = 'BUCKET',
'index.global.enabled' = 'true',
'read.utc-timezone' = 'false'
);
show create table myhudi.test_hudi3.hudi_datagen_incre_mysql_cdc;
insert into myhudi.test_hudi3.hudi_datagen_incre_mysql_cdc
select id,name,CAST(CURRENT_TIMESTAMP AS timestamp(3))
from cdc_test; {code}
* spark
{code:java}
select *
from
`test_hudi3`.hudi_datagen_incre_mysql_cdc {code}
* result
mysql
!image-2023-05-16-11-32-09-928.png!
hudi:
!image-2023-05-16-11-31-57-485.png!
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
