[
https://issues.apache.org/jira/browse/HUDI-9540?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18013656#comment-18013656
]
Hans Eschbaum commented on HUDI-9540:
-------------------------------------
I managed to reproduce this locally like this:
SET 'execution.runtime-mode' = 'batch';
SET 'sql-client.execution.result-mode' = 'tableau';
DROP TABLE IF EXISTS t_mor;
CREATE TABLE t_mor (
ts BIGINT,
uuid STRING PRIMARY KEY NOT ENFORCED,
rider STRING,
driver STRING,
fare DOUBLE,
city STRING
) PARTITIONED BY (city)
WITH (
'connector' = 'hudi',
'path' = 'file:///data/hudi_mor_test',
'table.type' = 'MERGE_ON_READ',
'record.key.field' = 'uuid',
'precombine.field' = 'ts',
'metadata.enabled' = 'false',
'index.key.field' = 'uuid',
'clean.retain_commits' = '2'
);
-- First delta commit
INSERT INTO t_mor VALUES
(1695159649001,'id-0001','r1','d1',10.0,'sf'),
(1695159649002,'id-0002','r2','d2',20.0,'sf'),
(1695159649003,'id-0003','r3','d3',30.0,'sf'),
(1695159649004,'id-0004','r4','d4',40.0,'sf'),
(1695159649005,'id-0005','r5','d5',50.0,'sf'),
(1695159649006,'id-0006','r6','d6',60.0,'nyc'),
(1695159649007,'id-0007','r7','d7',70.0,'nyc'),
(1695159649008,'id-0008','r8','d8',80.0,'nyc'),
(1695159649009,'id-0009','r9','d9',90.0,'nyc'),
(1695159649010,'id-0010','r10','d10',100.0,'nyc');
-- Second delta commit
INSERT INTO t_mor VALUES
(1695159649011,'id-0011','r11','d11',110.0,'sf'),
(1695159649012,'id-0012','r12','d12',120.0,'sf'),
(1695159649013,'id-0013','r13','d13',130.0,'sf'),
(1695159649014,'id-0014','r14','d14',140.0,'sf'),
(1695159649015,'id-0015','r15','d15',150.0,'sf'),
(1695159649016,'id-0016','r16','d16',160.0,'nyc'),
(1695159649017,'id-0017','r17','d17',170.0,'nyc'),
(1695159649018,'id-0018','r18','d18',180.0,'nyc'),
(1695159649019,'id-0019','r19','d19',190.0,'nyc'),
(1695159649020,'id-0020','r20','d20',200.0,'nyc'),
(1695159649021,'id-0021','r21','d21',210.0,'la');
-- Third delta commit
INSERT INTO t_mor VALUES
(1695159649022,'id-0022','r11','d11',110.0,'sf'),
(1695159649023,'id-0023','r12','d12',120.0,'sf'),
(1695159649024,'id-0024','r13','d13',130.0,'sf'),
(1695159649025,'id-0025','r14','d14',140.0,'sf'),
(1695159649026,'id-0026','r15','d15',150.0,'sf'),
(1695159649027,'id-0027','r16','d16',160.0,'nyc'),
(1695159649028,'id-0028','r17','d17',170.0,'nyc'),
(1695159649029,'id-0029','r18','d18',180.0,'nyc'),
(1695159649030,'id-0030','r19','d19',190.0,'nyc'),
(1695159649031,'id-0031','r20','d20',200.0,'nyc'),
(1695159649032,'id-0032','r21','d21',210.0,'la');
-- Fourth delta commit
INSERT INTO t_mor VALUES
(1695159649023,'id-0023','r11','d11',110.0,'sf'),
(1695159649024,'id-0024','r12','d12',120.0,'sf'),
(1695159649025,'id-0025','r13','d13',130.0,'sf'),
(1695159649026,'id-0026','r14','d14',140.0,'sf'),
(1695159649027,'id-0027','r15','d15',150.0,'sf'),
(1695159649028,'id-0028','r16','d16',160.0,'nyc'),
(1695159649029,'id-0029','r17','d17',170.0,'nyc'),
(1695159649030,'id-0030','r18','d18',180.0,'nyc'),
(1695159649031,'id-0031','r19','d19',190.0,'nyc'),
(1695159649032,'id-0032','r20','d20',200.0,'nyc'),
(1695159649033,'id-0033','r21','d21',210.0,'la');
-- Fifth delta commit
INSERT INTO t_mor VALUES
(1695159649034,'id-0011','r11','d11',110.0,'sf'),
(1695159649035,'id-0012','r12','d12',120.0,'sf'),
(1695159649036,'id-0013','r13','d13',130.0,'sf'),
(1695159649037,'id-0014','r14','d14',140.0,'sf'),
(1695159649038,'id-0015','r15','d15',150.0,'sf'),
(1695159649039,'id-0016','r16','d16',160.0,'nyc'),
(1695159649040,'id-0017','r17','d17',170.0,'nyc'),
(1695159649041,'id-0018','r18','d18',180.0,'nyc'),
(1695159649042,'id-0019','r19','d19',190.0,'nyc'),
(1695159649043,'id-0020','r20','d20',200.0,'nyc'),
(1695159649044,'id-0021','r21','d21',210.0,'la');
-- Sixth delta commit
INSERT INTO t_mor VALUES
(1695159649045,'id-0034','r11','d11',110.0,'sf'),
(1695159649046,'id-0035','r12','d12',120.0,'sf'),
(1695159649047,'id-0036','r13','d13',130.0,'sf'),
(1695159649048,'id-0037','r14','d14',140.0,'sf'),
(1695159649049,'id-0038','r15','d15',150.0,'sf'),
(1695159649050,'id-0039','r16','d16',160.0,'nyc'),
(1695159649051,'id-0040','r17','d17',170.0,'nyc'),
(1695159649052,'id-0041','r18','d18',180.0,'nyc'),
(1695159649053,'id-0042','r19','d19',190.0,'nyc'),
(1695159649054,'id-0043','r20','d20',200.0,'nyc'),
(1695159649055,'id-0044','r21','d21',210.0,'la');
-- Seventh delta commit
INSERT INTO t_mor VALUES
(1695159649056,'id-0045','r11','d11',110.0,'sf'),
(1695159649057,'id-0046','r12','d12',120.0,'sf'),
(1695159649058,'id-0047','r13','d13',130.0,'sf'),
(1695159649059,'id-0048','r14','d14',140.0,'sf'),
(1695159649060,'id-0049','r15','d15',150.0,'sf'),
(1695159649061,'id-0050','r16','d16',160.0,'nyc'),
(1695159649062,'id-0051','r17','d17',170.0,'nyc'),
(1695159649063,'id-0052','r18','d18',180.0,'nyc'),
(1695159649064,'id-0053','r19','d19',190.0,'nyc'),
(1695159649065,'id-0054','r20','d20',200.0,'nyc'),
(1695159649066,'id-0055','r21','d21',210.0,'la');
snap = (
spark.read.format("hudi")
.option("hoodie.metadata.enable", "false")
.option("hoodie.metadata.index.partition.stats.enable", "false")
.option("hoodie.metadata.index.column.stats.enable", "false")
.load(PATH)
).cache()
snap_cnt = snap.count()
incr = (
spark.read.format("hudi")
.option("hoodie.datasource.query.type", "incremental")
.option("hoodie.datasource.read.begin.instanttime", "000")
.option("hoodie.metadata.enable", "false")
.option("hoodie.metadata.index.partition.stats.enable", "false")
.option("hoodie.metadata.index.column.stats.enable", "false")
.load(PATH)
).cache()
incr_cnt = incr.count()
print(f"SNAPSHOT_COUNT=\{snap_cnt}")
print(f"INCREMENTAL_COUNT=\{incr_cnt}")
SNAPSHOT_COUNT=55
INCREMENTAL_COUNT=33
The incremental query is not showing the last two commits, id-0034 to id-0055.
timeline:
13.08.2025 12:58 8 .20250813095856870.deltacommit.inflight.crc
13.08.2025 12:58 8 .20250813095856870.deltacommit.requested.crc
13.08.2025 12:58 44
.20250813095856870_20250813095858533.deltacommit.crc
13.08.2025 12:59 8 .20250813095904671.deltacommit.inflight.crc
13.08.2025 12:59 8 .20250813095904671.deltacommit.requested.crc
13.08.2025 12:59 44
.20250813095904671_20250813095906629.deltacommit.crc
13.08.2025 12:59 8 .20250813095912307.deltacommit.inflight.crc
13.08.2025 12:59 8 .20250813095912307.deltacommit.requested.crc
13.08.2025 12:59 44
.20250813095912307_20250813095914418.deltacommit.crc
13.08.2025 12:59 8 .20250813095920001.deltacommit.inflight.crc
13.08.2025 12:59 8 .20250813095920001.deltacommit.requested.crc
13.08.2025 12:59 44
.20250813095920001_20250813095922227.deltacommit.crc
13.08.2025 12:59 8 .20250813095927702.deltacommit.inflight.crc
13.08.2025 12:59 8 .20250813095927702.deltacommit.requested.crc
13.08.2025 12:59 44
.20250813095927702_20250813095930131.deltacommit.crc
13.08.2025 12:59 8 .20250813095930511.compaction.inflight.crc
13.08.2025 12:59 40 .20250813095930511.compaction.requested.crc
13.08.2025 12:59 44
.20250813095930511_20250813095934925.commit.crc
13.08.2025 12:59 8 .20250813095934482.deltacommit.inflight.crc
13.08.2025 12:59 8 .20250813095934482.deltacommit.requested.crc
13.08.2025 12:59 44
.20250813095934482_20250813095937806.deltacommit.crc
13.08.2025 12:59 8 .20250813095942456.deltacommit.inflight.crc
13.08.2025 12:59 8 .20250813095942456.deltacommit.requested.crc
13.08.2025 12:59 44
.20250813095942456_20250813095944845.deltacommit.crc
13.08.2025 12:59 36 .20250813095946580.clean.inflight.crc
13.08.2025 12:59 36 .20250813095946580.clean.requested.crc
13.08.2025 12:59 40
.20250813095946580_20250813095948812.clean.crc
13.08.2025 12:58 0 20250813095856870.deltacommit.inflight
13.08.2025 12:58 0 20250813095856870.deltacommit.requested
13.08.2025 12:58 4 295
20250813095856870_20250813095858533.deltacommit
13.08.2025 12:59 0 20250813095904671.deltacommit.inflight
13.08.2025 12:59 0 20250813095904671.deltacommit.requested
13.08.2025 12:59 4 560
20250813095904671_20250813095906629.deltacommit
13.08.2025 12:59 0 20250813095912307.deltacommit.inflight
13.08.2025 12:59 0 20250813095912307.deltacommit.requested
13.08.2025 12:59 4 560
20250813095912307_20250813095914418.deltacommit
13.08.2025 12:59 0 20250813095920001.deltacommit.inflight
13.08.2025 12:59 0 20250813095920001.deltacommit.requested
13.08.2025 12:59 4 560
20250813095920001_20250813095922227.deltacommit
13.08.2025 12:59 0 20250813095927702.deltacommit.inflight
13.08.2025 12:59 0 20250813095927702.deltacommit.requested
13.08.2025 12:59 4 560
20250813095927702_20250813095930131.deltacommit
13.08.2025 12:59 0 20250813095930511.compaction.inflight
13.08.2025 12:59 3 804 20250813095930511.compaction.requested
13.08.2025 12:59 4 316 20250813095930511_20250813095934925.commit
13.08.2025 12:59 0 20250813095934482.deltacommit.inflight
13.08.2025 12:59 0 20250813095934482.deltacommit.requested
13.08.2025 12:59 4 560
20250813095934482_20250813095937806.deltacommit
13.08.2025 12:59 0 20250813095942456.deltacommit.inflight
13.08.2025 12:59 0 20250813095942456.deltacommit.requested
13.08.2025 12:59 4 560
20250813095942456_20250813095944845.deltacommit
13.08.2025 12:59 3 208 20250813095946580.clean.inflight
13.08.2025 12:59 3 208 20250813095946580.clean.requested
13.08.2025 12:59 3 695 20250813095946580_20250813095948812.clean
Tested this in a docker environment, with the following packages:
https://repo1.maven.org/maven2/org/apache/hudi/hudi-flink1.20-bundle/1.0.2/hudi-flink1.20-bundle-1.0.2.jar
https://repo1.maven.org/maven2/org/apache/hudi/hudi-spark3.5-bundle_2.12/1.0.2/hudi-spark3.5-bundle_2.12-1.0.2.jar
https://repo1.maven.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar
It seems to me that the clean is somehow responsible, when I set
'clean.retain_commits' = '18' then with this setup the counts matched, but
eventually when the clean is performed, after adding enough commits it starts
happening again. Let me know if I can provide any additional information.
> Incremental Reads Issue With Flink
> ----------------------------------
>
> Key: HUDI-9540
> URL: https://issues.apache.org/jira/browse/HUDI-9540
> Project: Apache Hudi
> Issue Type: Bug
> Affects Versions: 1.0.2
> Reporter: Hans Eschbaum
> Priority: Major
>
> Hey!
> There seems to be an issue with incremental reads with MoR tables in Flink.
>
> When I read via PySpark with, ie.
> {code:java}
> read_options = {
> 'hoodie.datasource.query.type': 'incremental',
> 'hoodie.datasource.read.begin.instanttime': '0',
> }{code}
> Some of the records do not show up that do show up with snapshot reads.
> I think this issue is likely related to the differences between MoR writes
> between Spark and Flink, where Spark always puts inserts into the base file,
> but Flink puts the inserts into the log files as well. So for some reason
> some or all of these inserts that only exist in the log files and haven't
> been compacted into base files are ignored during incremental reads.
> (My data has no deletes)
> Hudi version: 1.0.2
> Flink Version: 1.20.0
> Storage: S3
> (Writes are via AWS Managed Flink, reads are via EMR Serverless)
> To reproduce:
> 1. Do some writes via Flink to a Hudi sink
> 2. Compare an incremental read to a snapshot read, some of the records go
> missing.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
