[SUPPORT] Custom payload records ordering is different for incremental and snapshot queries
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
I implemented a custom payload based on HoodieRecordPayload.java, but there were problems. When I use incremental queries, record_time is the value of the incremental payload (incorrect). When running a snapshot query, record_time is an old value (correct), which does not meet my expectations. Does the payload obtained by incremental queries differ from the snapshot query results? Please help me
```java
/* Omitted content */
public class CustomPayload extends OverwriteWithLatestAvroPayload {
/* Omitted content */
@Override
public Option combineAndGetUpdateValue(IndexedRecord currentValue, Schema schema, Properties properties) throws IOException {
if (recordBytes.length == 0) {
return Option.empty();
}
GenericRecord incomingRecord = HoodieAvroUtils.bytesToAvro(recordBytes, schema);
if (!needUpdatingPersistedRecord(currentValue, incomingRecord, properties)) {
return Option.of(currentValue);
}
/*custom code*/
if (((GenericRecord) currentValue).get("record_time") != null) {
incomingRecord.put("record_time", ((GenericRecord) currentValue).get("record_time"));
}
eventTime = updateEventTime(incomingRecord, properties);
return isDeleteRecord(incomingRecord) ? Option.empty() : Option.of(incomingRecord);
}
/* Omitted content */
protected boolean needUpdatingPersistedRecord(IndexedRecord currentValue,
IndexedRecord incomingRecord, Properties properties) {
/* Omitted content */
return (((Comparable) persistedOrderingVal).compareTo(incomingOrderingVal) < 0) && (
((GenericRecord) currentValue).get("valid").equals(true) || ((GenericRecord) incomingRecord).get("valid").equals(true)) && (
((GenericRecord) currentValue).get("content_md5") == null
|| !((GenericRecord) currentValue).get("content_md5").equals(((GenericRecord) incomingRecord).get("content_md5"))
);
}
}
```
**Expected behavior**
snapshot query
```shell
>>> spark.read.format("hudi").load("*****").filter("*********").show(truncate=False)
+-------------------+----------------------------+----------------------------------------------------------------
+----------------------+------------------------------------------------------------------------+----
|_hoodie_commit_time|_hoodie_commit_seqno |_hoodie_record_key
|_hoodie_partition_path|_hoodie_file_name
|valid|update_time |record_time |content_md5
|id |
+-------------------+----------------------------+----------------------------------------------------------------
+----------------------+------------------------------------------------------------------------+----
|20230324152704225 |20230324152704225_13_2563860|7306da3dd0c41ff504447981c4e850949db69524154c0c5bf85e62758babf3cc
| |00000013-ea30-4f9d-9704-e4f82fceb940-0 |
false|2023-03-24 14:48:09|2023-03-01 21:40:42|df3c2a9f8eaf5b8eec26b363cc67003f
|7306da3dd0c41ff504447981c4e850949db69524154c0c5bf85e62758babf3cc|
```
incremental query
```shell
>>> df = spark.read.format("hudi").options(**{'hoodie.datasource.query.type': "incremental", "hoodie.datasource.read.begin.instanttime": '20230324151944584'}).load("******")
>>> df.filter("*********").show(truncate=False)
+-------------------+----------------------------+----------------------------------------------------------------
+----------------------+------------------------------------------------------------------------+----
|_hoodie_commit_time|_hoodie_commit_seqno |_hoodie_record_key
|_hoodie_partition_path|_hoodie_file_name
|valid|update_time |record_time |content_md5
|id |
+-------------------+----------------------------+----------------------------------------------------------------
+----------------------+------------------------------------------------------------------------+----
|20230324152704225 |20230324152704225_13_2563860|7306da3dd0c41ff504447981c4e850949db69524154c0c5bf85e62758babf3cc
| |00000013-ea30-4f9d-9704-e4f82fceb940-0 |
false|2023-03-24 14:48:09|2023-03-24 14:48:09|df3c2a9f8eaf5b8eec26b363cc67003f
|7306da3dd0c41ff504447981c4e850949db69524154c0c5bf85e62758babf3cc|
```
I read kafka from spark structured streaming, write it to hudi use foreachBatch. There are two time fields, **update_time** and **record_time**. By default, the values of the two time fields are the same. Combined with the custom payload, it is implemented to determine whether to update. If updated, the persistent payload record_time overwrite new payload , I think in incremental queries, record_time should be the same as the snapshot query, but the result is different from what I expected. Why is this, pls help me
**Environment Description**
* Hudi version : 0.12.1
* Spark version : 3.2.2
* Hive version : 2.1.1
* Hadoop version : 3.0.0
* Storage (HDFS/S3/GCS..) : HDFS
* Running on Docker? (yes/no) : no
**Additional context**
Add any other context about the problem here.
**Stacktrace**
```Add the stacktrace of the error.```
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with HoodieRecordPayload.java and the custom combineAndGetUpdateValue and needUpdatingPersistedRecord methods shown in the issue. Reproduce the snapshot and incremental Spark reads using the supplied Hudi 0.12.1 and Spark 3.2.2 settings, then trace how each query obtains record_time. Done means the difference is explained and the expected record_time behavior is documented or corrected.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, spark
- Domain
- data-engineering, stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100