You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "robbik (via GitHub)" <gi...@apache.org> on 2025/03/07 08:29:39 UTC
[I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
robbik opened a new issue, #12931:
URL: https://github.com/apache/hudi/issues/12931
**_Tips before filing an issue_**
- Have you gone through our [FAQs](https://hudi.apache.org/learn/faq/)?
- Join the mailing list to engage in conversations and get faster support at dev-subscribe@hudi.apache.org.
- If you have triaged this as a bug, then file an [issue](https://issues.apache.org/jira/projects/HUDI/issues) directly.
**Describe the problem you faced**
I am doing upsert to existing hudi MOR table. When doing the upsert, an error / failure has occured with message java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided.
**To Reproduce**
I made simple java unit test to simulate this using my existing record_index file and new data to be upserted.
```
package org.apache.hudi.io.hfile;
import bav.datalake.etl.spark.util.Files;
import org.apache.hudi.common.util.io.ByteBufferBackedInputStream;
import org.apache.hudi.io.ByteArraySeekableDataInputStream;
import org.junit.jupiter.api.Test;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
class HFileReaderImplTest {
@Test
void test() throws Exception {
final byte[] contents = Files.toBytes("/abc/.hoodie/metadata/record_index/.record-index-0028-0_20250226192018634.log.1_102-49-7606");
final UTF8StringKey LASTKEY = new UTF8StringKey("hfile.LASTKEY");
try (HFileReaderImpl reader = new HFileReaderImpl(new ByteArraySeekableDataInputStream(new ByteBufferBackedInputStream(contents, 8036, 3944824)), 3944824)) {
assertThat(reader.seekTo(), is(true));
assertThat(new Key(reader.getMetaInfo(LASTKEY).get()).getContentInString(), is("OurBranchID:9999,AccountID:9999NLP00001,LoanSeries:1,InstallmentNo:6"));
System.out.println(reader.seekTo(new UTF8StringKey("OurBranchID:0324,AccountID:0324NT700001,LoanSeries:1,InstallmentNo:3")));
System.out.println(reader.seekTo(new UTF8StringKey("OurBranchID:0324,AccountID:0324NT700001,LoanSeries:1,InstallmentNo:30")));
}
}
}
```
I attach the record index file (zipped).
**Expected behavior**
No error occurred and the upsert is successful.
**Environment Description**
* Hudi version : 1.0.1
* Spark version : 3.5.2 (scala 2.12)
* Hive version : -
* Hadoop version : 2.7.1
* Storage (HDFS/S3/GCS..) : S3, Local file system (no difference)
* Running on Docker? (yes/no) : yes and no (no difference)
[.record-index-0028-0_20250226192018634.log.zip](https://github.com/user-attachments/files/19123040/default.record-index-0028-0_20250226192018634.log.zip)
**Stacktrace**
```
java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided. key=UTF8StringKey{OurBranchID:0324,AccountID:0324NT700001,LoanSeries:1,InstallmentNo:30} cursor=HFilePosition{offset=3485859, keyValue=Option{val=KeyValue{key=Key{OurBranchID:0324,AccountID:0324REG00002,LoanSeries:1,InstallmentNo:5}}}}
at org.apache.hudi.io.hfile.HFileReaderImpl.seekTo(HFileReaderImpl.java:173)
at org.apache.hudi.io.hfile.HFileReaderImplTest.test(HFileReaderImplTest.java:29)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "rangareddy (via GitHub)" <gi...@apache.org>.
rangareddy commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2743244267
Hi @robbik
Thank you for sharing the code again. I will test it and get back to you regarding the action items for this issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "rangareddy (via GitHub)" <gi...@apache.org>.
rangareddy commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2709905580
Hi @robbik
Could you please share the sample reproducible code to test my end?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "robbik (via GitHub)" <gi...@apache.org>.
robbik commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2710789737
Here I attach the source code
[src.zip](https://github.com/user-attachments/files/19164567/src.zip)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "robbik (via GitHub)" <gi...@apache.org>.
robbik commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2819900598
and update on this issue? it is a blocker for me to use hudi 1.0 with row level indexing in production environment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "robbik (via GitHub)" <gi...@apache.org>.
robbik commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2707008191
Hi,
I did initial data insert using Spark and Hudi from one database to S3 using EMR. The S3 path was empty before doing the insert.
After the insert was successful, I did delete + upsert using Spark and Hudi from the same table (the table had some removed, updated, and new rows) to the same S3 path. Then the upsert was failed with error above but the delete was successful.
I retried several times but the result (error) was consistent even if I skipped the delete part.
I debugged the Spark job and saw that the lookup record was fine for other record index files but error for that particular record index file.
I debugged deeper into Hudi and found that in that record index file, at least one index block key pointed to different key in data block. Not sure if this is because of delete operation or this is expected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "rangareddy (via GitHub)" <gi...@apache.org>.
rangareddy commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2717687846
Hi @robbik
I used the following code to reproduce this issue, but I was not able to reproduce it.
```sh
export SPARK_VERSION=3.5
spark-shell --packages org.apache.hudi:hudi-spark$SPARK_VERSION-bundle_2.12:1.0.1 \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
--conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' \
--conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
--conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar'
```
```scala
import scala.collection.JavaConversions._
import org.apache.spark.sql.SaveMode._
import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.common.table.HoodieTableConfig._
import org.apache.hudi.config.HoodieWriteConfig._
import org.apache.hudi.keygen.constant.KeyGeneratorOptions._
import org.apache.hudi.common.model.HoodieRecord
import org.apache.hudi.config.HoodieCleanConfig
import org.apache.hudi.config.HoodieIndexConfig
import org.apache.hudi.config.HoodieCompactionConfig
import org.apache.hudi.common.config.HoodieMetadataConfig
import org.apache.hudi.hive.HiveSyncConfig
import org.apache.hudi.common.table.HoodieTableConfig
import org.apache.hudi.common.config.HoodieStorageConfig
import org.apache.hudi.index.HoodieIndex
import org.apache.hudi.common.model.HoodieTableType
import org.apache.hudi.sync.common.HoodieSyncConfig
import spark.implicits._
val tableName = "trips_table"
val basePath = "s3a://warehouse/trips_table"
val columns = Seq("ts","uuid","rider","driver","fare","city")
val data =
Seq((1695159649087L,"334e26e9-8355-45cc-97c6-c31daf0df330","rider-A","driver-K",19.10,"san_francisco"),
(1695091554788L,"e96c4396-3fad-413a-a942-4cb36106d721","rider-C","driver-M",27.70 ,"san_francisco"),
(1695046462179L,"9909a8b1-2d15-4d3d-8ec9-efc48c536a00","rider-D","driver-L",33.90 ,"san_francisco"),
(1695516137016L,"e3cf430c-889d-4015-bc98-59bdce1e530c","rider-F","driver-P",34.15,"sao_paulo" ),
(1695115999911L,"c8abbe79-8d89-47ea-b4ce-4d224bae5bfa","rider-J","driver-T",17.85,"chennai"));
val hudiOptions = Map(
"hoodie.write.record.merge.mode" -> "COMMIT_TIME_ORDERING",
"hoodie.datasource.write.partitionpath.field" -> "city",
"hoodie.table.name" -> tableName,
"hoodie.datasource.write.table.type" -> "MERGE_ON_READ",
"hoodie.metadata.enable" -> "true",
"hoodie.metadata.index.column.stats.enable" -> "true",
"hoodie.metadata.index.partition.stats.enable" -> "true",
"hoodie.metadata.index.bloom.filter.enable" -> "true",
"hoodie.metadata.record.index.enable" -> "true",
"hoodie.metadata.record.index.max.filegroup.count" -> "500000000",
"hoodie.metadata.record.index.min.filegroup.count" -> "100",
HoodieIndexConfig.INDEX_TYPE.key() -> HoodieIndex.IndexType.RECORD_INDEX.name(),
HoodieIndexConfig.RECORD_INDEX_UPDATE_PARTITION_PATH_ENABLE.key() -> "true",
HoodieIndexConfig.RECORD_INDEX_USE_CACHING.key() -> "true",
HoodieIndexConfig.BLOOM_INDEX_UPDATE_PARTITION_PATH_ENABLE.key() -> "true",
HoodieIndexConfig.BLOOM_INDEX_USE_CACHING.key() -> "true",
HoodieIndexConfig.BLOOM_INDEX_USE_METADATA.key() -> "true",
HoodieCleanConfig.CLEANER_POLICY.key() -> "KEEP_LATEST_COMMITS",
HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key() -> "10",
HoodieCompactionConfig.INLINE_COMPACT.key() -> "true",
HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key() -> "20",
HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT.key() -> "0",
HoodieMetadataConfig.ENABLE_OPTIMIZED_LOG_BLOCKS_SCAN.key() -> "true",
HoodieSyncConfig.META_SYNC_ENABLED.key() -> "false",
HiveSyncConfig.HIVE_SYNC_ENABLED.key() -> "false",
HoodieTableConfig.TYPE.key() -> HoodieTableType.MERGE_ON_READ.name(),
HoodieStorageConfig.PARQUET_COMPRESSION_CODEC_NAME.key() -> "snappy"
)
var insertsDF = spark.createDataFrame(data).toDF(columns:_*)
insertsDF.write.format("hudi").
options(hudiOptions).
mode(Overwrite).
save(basePath)
spark.read.format("hudi").load(basePath).show()
val deletesDF = spark.read.format("hudi").load(basePath).filter($"rider" === "rider-F")
deletesDF.write.format("hudi").
option("hoodie.datasource.write.operation", "delete").
options(hudiOptions).
mode(Append).
save(basePath)
spark.read.format("hudi").load(basePath).show()
val updatesDf = spark.read.format("hudi").load(basePath).filter($"rider" === "rider-D").withColumn("fare", col("fare") * 10)
updatesDf.write.format("hudi").
option("hoodie.datasource.write.operation", "upsert").
options(hudiOptions).
mode(Append).
save(basePath)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2819995211
Check out these two fixes: https://github.com/apache/hudi/pull/13166 and https://github.com/apache/hudi/pull/13179. Should be resolved in 1.0.2.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2788697449
@linliu-code Not sure if you are familiar with the Hbase APIs, it's great if you can give some help here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "robbik (via GitHub)" <gi...@apache.org>.
robbik commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2820362553
I have retested using the patches and it is OK now. Thank you
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "robbik (via GitHub)" <gi...@apache.org>.
robbik commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2788123410
or can you give me clue which one is wrong, is it the "seek" functionality in "HFileReaderImpl" or the structure of the row index file (first data block must match entry in index block)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "robbik (via GitHub)" <gi...@apache.org>.
robbik closed issue #12931: [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided
URL: https://github.com/apache/hudi/issues/12931
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "robbik (via GitHub)" <gi...@apache.org>.
robbik commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2707971955
and here is the full stack trace of error while doing upsert.
```
10:07:54,937 ERROR [org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner] Got exception when reading log file
java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided. key=UTF8StringKey{OurBranchID:0324,AccountID:0324NT700001,LoanSeries:1,InstallmentNo:30} cursor=HFilePosition{offset=3485859, keyValue=Option{val=KeyValue{key=Key{OurBranchID:0324,AccountID:0324REG00002,LoanSeries:1,InstallmentNo:5}}}}
at org.apache.hudi.io.hfile.HFileReaderImpl.seekTo(HFileReaderImpl.java:173) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.io.storage.HoodieNativeAvroHFileReader$RecordByKeyIterator.hasNext(HoodieNativeAvroHFileReader.java:404) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.util.collection.MappingIterator.hasNext(MappingIterator.java:39) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.util.collection.MappingIterator.hasNext(MappingIterator.java:39) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.util.collection.MappingIterator.hasNext(MappingIterator.java:39) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.processDataBlock(AbstractHoodieLogRecordScanner.java:633) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.processQueuedBlocksForInstant(AbstractHoodieLogRecordScanner.java:675) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.scanInternalV2(AbstractHoodieLogRecordScanner.java:595) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.scanInternal(AbstractHoodieLogRecordScanner.java:248) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner.scanByFullKeys(HoodieMergedLogRecordScanner.java:163) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieMetadataLogRecordReader.getAllRecordsByKeys(HoodieMetadataLogRecordReader.java:127) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieBackedTableMetadata.readAllLogRecords(HoodieBackedTableMetadata.java:477) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieBackedTableMetadata.lookupAllKeysFromFileSlice(HoodieBackedTableMetadata.java:455) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieBackedTableMetadata.lambda$getAllRecordsByKeys$61b9eeed$1(HoodieBackedTableMetadata.java:327) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.function.FunctionWrapper.lambda$throwingMapWrapper$0(FunctionWrapper.java:38) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) [?:?]
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625) [?:?]
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) [?:?]
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) [?:?]
at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:960) [?:?]
at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:934) [?:?]
at java.util.stream.AbstractTask.compute(AbstractTask.java:327) [?:?]
at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:754) [?:?]
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:373) [?:?]
at java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1182) [?:?]
at java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1655) [?:?]
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1622) [?:?]
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:165) [?:?]
10:07:55,000 ERROR [org.apache.spark.executor.Executor] Exception in task 28.0 in stage 2.0 (TID 553)
org.apache.hudi.exception.HoodieException: org.apache.hudi.exception.HoodieException: Error occurs when executing map
at jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[?:?]
at jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:77) ~[?:?]
at jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) ~[?:?]
at java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:499) ~[?:?]
at java.lang.reflect.Constructor.newInstance(Constructor.java:480) ~[?:?]
at java.util.concurrent.ForkJoinTask.getThrowableException(ForkJoinTask.java:562) ~[?:?]
at java.util.concurrent.ForkJoinTask.reportException(ForkJoinTask.java:591) ~[?:?]
at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:689) ~[?:?]
at java.util.stream.ReduceOps$ReduceOp.evaluateParallel(ReduceOps.java:927) ~[?:?]
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233) ~[?:?]
at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682) ~[?:?]
at org.apache.hudi.common.engine.HoodieLocalEngineContext.map(HoodieLocalEngineContext.java:90) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieBackedTableMetadata.getAllRecordsByKeys(HoodieBackedTableMetadata.java:322) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.BaseTableMetadata.readRecordIndex(BaseTableMetadata.java:299) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.index.SparkMetadataTableRecordIndex$RecordIndexFileGroupLookupFunction.call(SparkMetadataTableRecordIndex.java:166) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.index.SparkMetadataTableRecordIndex$RecordIndexFileGroupLookupFunction.call(SparkMetadataTableRecordIndex.java:153) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsToPair$1(JavaRDDLike.scala:186) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:858) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:858) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDD.iterator(RDD.scala:331) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:104) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:54) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.scheduler.Task.run(Task.scala:141) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:620) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64) ~[spark-common-utils_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61) ~[spark-common-utils_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:623) [spark-core_2.12-3.5.2.jar:3.5.2]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
at java.lang.Thread.run(Thread.java:833) [?:?]
Caused by: org.apache.hudi.exception.HoodieException: Error occurs when executing map
at org.apache.hudi.common.function.FunctionWrapper.lambda$throwingMapWrapper$0(FunctionWrapper.java:40) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) ~[?:?]
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625) ~[?:?]
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[?:?]
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[?:?]
at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:960) ~[?:?]
at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:934) ~[?:?]
at java.util.stream.AbstractTask.compute(AbstractTask.java:327) ~[?:?]
at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:754) ~[?:?]
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:373) ~[?:?]
at java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1182) ~[?:?]
at java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1655) ~[?:?]
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1622) ~[?:?]
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:165) ~[?:?]
Caused by: org.apache.hudi.exception.HoodieException: Exception when reading log file
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.scanInternalV2(AbstractHoodieLogRecordScanner.java:604) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.scanInternal(AbstractHoodieLogRecordScanner.java:248) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner.scanByFullKeys(HoodieMergedLogRecordScanner.java:163) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieMetadataLogRecordReader.getAllRecordsByKeys(HoodieMetadataLogRecordReader.java:127) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieBackedTableMetadata.readAllLogRecords(HoodieBackedTableMetadata.java:477) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieBackedTableMetadata.lookupAllKeysFromFileSlice(HoodieBackedTableMetadata.java:455) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieBackedTableMetadata.lambda$getAllRecordsByKeys$61b9eeed$1(HoodieBackedTableMetadata.java:327) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.function.FunctionWrapper.lambda$throwingMapWrapper$0(FunctionWrapper.java:38) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) ~[?:?]
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625) ~[?:?]
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[?:?]
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[?:?]
at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:960) ~[?:?]
at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:934) ~[?:?]
at java.util.stream.AbstractTask.compute(AbstractTask.java:327) ~[?:?]
at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:754) ~[?:?]
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:373) ~[?:?]
at java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1182) ~[?:?]
at java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1655) ~[?:?]
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1622) ~[?:?]
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:165) ~[?:?]
Caused by: java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided. key=UTF8StringKey{OurBranchID:0324,AccountID:0324NT700001,LoanSeries:1,InstallmentNo:30} cursor=HFilePosition{offset=3485859, keyValue=Option{val=KeyValue{key=Key{OurBranchID:0324,AccountID:0324REG00002,LoanSeries:1,InstallmentNo:5}}}}
at org.apache.hudi.io.hfile.HFileReaderImpl.seekTo(HFileReaderImpl.java:173) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.io.storage.HoodieNativeAvroHFileReader$RecordByKeyIterator.hasNext(HoodieNativeAvroHFileReader.java:404) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.util.collection.MappingIterator.hasNext(MappingIterator.java:39) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.util.collection.MappingIterator.hasNext(MappingIterator.java:39) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.util.collection.MappingIterator.hasNext(MappingIterator.java:39) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.processDataBlock(AbstractHoodieLogRecordScanner.java:633) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.processQueuedBlocksForInstant(AbstractHoodieLogRecordScanner.java:675) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.scanInternalV2(AbstractHoodieLogRecordScanner.java:595) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.scanInternal(AbstractHoodieLogRecordScanner.java:248) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner.scanByFullKeys(HoodieMergedLogRecordScanner.java:163) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieMetadataLogRecordReader.getAllRecordsByKeys(HoodieMetadataLogRecordReader.java:127) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieBackedTableMetadata.readAllLogRecords(HoodieBackedTableMetadata.java:477) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieBackedTableMetadata.lookupAllKeysFromFileSlice(HoodieBackedTableMetadata.java:455) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieBackedTableMetadata.lambda$getAllRecordsByKeys$61b9eeed$1(HoodieBackedTableMetadata.java:327) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.function.FunctionWrapper.lambda$throwingMapWrapper$0(FunctionWrapper.java:38) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) ~[?:?]
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625) ~[?:?]
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[?:?]
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[?:?]
at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:960) ~[?:?]
at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:934) ~[?:?]
at java.util.stream.AbstractTask.compute(AbstractTask.java:327) ~[?:?]
at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:754) ~[?:?]
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:373) ~[?:?]
at java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1182) ~[?:?]
at java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1655) ~[?:?]
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1622) ~[?:?]
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:165) ~[?:?]
```
```
org.apache.hudi.exception.HoodieUpsertException: Failed to upsert for commit time 20250308100148957
at org.apache.hudi.table.action.commit.BaseWriteHelper.write(BaseWriteHelper.java:64) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.table.action.deltacommit.SparkUpsertDeltaCommitActionExecutor.execute(SparkUpsertDeltaCommitActionExecutor.java:45) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.table.HoodieSparkMergeOnReadTable.upsert(HoodieSparkMergeOnReadTable.java:98) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.table.HoodieSparkMergeOnReadTable.upsert(HoodieSparkMergeOnReadTable.java:88) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.client.SparkRDDWriteClient.upsert(SparkRDDWriteClient.java:132) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.DataSourceUtils.doWriteOperation(DataSourceUtils.java:227) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.HoodieSparkSqlWriterInternal.liftedTree1$1(HoodieSparkSqlWriter.scala:517) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.HoodieSparkSqlWriterInternal.writeInternal(HoodieSparkSqlWriter.scala:515) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.HoodieSparkSqlWriterInternal.$anonfun$write$1(HoodieSparkSqlWriter.scala:192) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:125) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:201) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:108) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:900) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:66) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.adapter.BaseSpark3Adapter.sqlExecutionWithNewExecutionId(BaseSpark3Adapter.scala:105) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.HoodieSparkSqlWriterInternal.write(HoodieSparkSqlWriter.scala:214) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.HoodieSparkSqlWriter$.write(HoodieSparkSqlWriter.scala:129) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.DefaultSource.createRelation(DefaultSource.scala:170) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:48) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:75) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:73) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.command.ExecutedCommandExec.executeCollect(commands.scala:84) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.$anonfun$applyOrElse$1(QueryExecution.scala:107) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:125) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:201) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:108) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:900) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:66) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:107) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:98) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:461) ~[spark-catalyst_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:76) ~[spark-sql-api_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:461) ~[spark-catalyst_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:32) ~[spark-catalyst_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267) ~[spark-catalyst_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263) ~[spark-catalyst_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:32) ~[spark-catalyst_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:32) ~[spark-catalyst_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:437) ~[spark-catalyst_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.QueryExecution.eagerlyExecuteCommands(QueryExecution.scala:98) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.QueryExecution.commandExecuted$lzycompute(QueryExecution.scala:85) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.QueryExecution.commandExecuted(QueryExecution.scala:83) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.execution.QueryExecution.assertCommandExecuted(QueryExecution.scala:142) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:859) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:388) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.DataFrameWriter.saveInternal(DataFrameWriter.scala:361) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:248) ~[spark-sql_2.12-3.5.2.jar:3.5.2]
...
Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: Task 28 in stage 2.0 failed 1 times, most recent failure: Lost task 28.0 in stage 2.0 (TID 553) (10.255.255.254 executor driver): org.apache.hudi.exception.HoodieException: org.apache.hudi.exception.HoodieException: Error occurs when executing map
at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.base/java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:499)
at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:480)
at java.base/java.util.concurrent.ForkJoinTask.getThrowableException(ForkJoinTask.java:562)
at java.base/java.util.concurrent.ForkJoinTask.reportException(ForkJoinTask.java:591)
at java.base/java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:689)
at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateParallel(ReduceOps.java:927)
at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233)
at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682)
at org.apache.hudi.common.engine.HoodieLocalEngineContext.map(HoodieLocalEngineContext.java:90)
at org.apache.hudi.metadata.HoodieBackedTableMetadata.getAllRecordsByKeys(HoodieBackedTableMetadata.java:322)
at org.apache.hudi.metadata.BaseTableMetadata.readRecordIndex(BaseTableMetadata.java:299)
at org.apache.hudi.index.SparkMetadataTableRecordIndex$RecordIndexFileGroupLookupFunction.call(SparkMetadataTableRecordIndex.java:166)
at org.apache.hudi.index.SparkMetadataTableRecordIndex$RecordIndexFileGroupLookupFunction.call(SparkMetadataTableRecordIndex.java:153)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsToPair$1(JavaRDDLike.scala:186)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:858)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:858)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:331)
at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:104)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:54)
at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166)
at org.apache.spark.scheduler.Task.run(Task.scala:141)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:620)
at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64)
at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:623)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.apache.hudi.exception.HoodieException: Error occurs when executing map
at org.apache.hudi.common.function.FunctionWrapper.lambda$throwingMapWrapper$0(FunctionWrapper.java:40)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625)
at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
at java.base/java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:960)
at java.base/java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:934)
at java.base/java.util.stream.AbstractTask.compute(AbstractTask.java:327)
at java.base/java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:754)
at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:373)
at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1182)
at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1655)
at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1622)
at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:165)
Caused by: org.apache.hudi.exception.HoodieException: Exception when reading log file
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.scanInternalV2(AbstractHoodieLogRecordScanner.java:604)
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.scanInternal(AbstractHoodieLogRecordScanner.java:248)
at org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner.scanByFullKeys(HoodieMergedLogRecordScanner.java:163)
at org.apache.hudi.metadata.HoodieMetadataLogRecordReader.getAllRecordsByKeys(HoodieMetadataLogRecordReader.java:127)
at org.apache.hudi.metadata.HoodieBackedTableMetadata.readAllLogRecords(HoodieBackedTableMetadata.java:477)
at org.apache.hudi.metadata.HoodieBackedTableMetadata.lookupAllKeysFromFileSlice(HoodieBackedTableMetadata.java:455)
at org.apache.hudi.metadata.HoodieBackedTableMetadata.lambda$getAllRecordsByKeys$61b9eeed$1(HoodieBackedTableMetadata.java:327)
at org.apache.hudi.common.function.FunctionWrapper.lambda$throwingMapWrapper$0(FunctionWrapper.java:38)
... 13 more
Caused by: java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided. key=UTF8StringKey{OurBranchID:0324,AccountID:0324NT700001,LoanSeries:1,InstallmentNo:30} cursor=HFilePosition{offset=3485859, keyValue=Option{val=KeyValue{key=Key{OurBranchID:0324,AccountID:0324REG00002,LoanSeries:1,InstallmentNo:5}}}}
at org.apache.hudi.io.hfile.HFileReaderImpl.seekTo(HFileReaderImpl.java:173)
at org.apache.hudi.io.storage.HoodieNativeAvroHFileReader$RecordByKeyIterator.hasNext(HoodieNativeAvroHFileReader.java:404)
at org.apache.hudi.common.util.collection.MappingIterator.hasNext(MappingIterator.java:39)
at org.apache.hudi.common.util.collection.MappingIterator.hasNext(MappingIterator.java:39)
at org.apache.hudi.common.util.collection.MappingIterator.hasNext(MappingIterator.java:39)
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.processDataBlock(AbstractHoodieLogRecordScanner.java:633)
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.processQueuedBlocksForInstant(AbstractHoodieLogRecordScanner.java:675)
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.scanInternalV2(AbstractHoodieLogRecordScanner.java:595)
... 20 more
```
and the driver stack trace:
```
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2856) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2792) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2791) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62) ~[scala-library-2.12.18.jar:?]
at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55) ~[scala-library-2.12.18.jar:?]
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49) ~[scala-library-2.12.18.jar:?]
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2791) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:1247) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:1247) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at scala.Option.foreach(Option.scala:407) ~[scala-library-2.12.18.jar:?]
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1247) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:3060) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2994) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2983) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:989) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2393) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2414) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2433) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2458) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDD.$anonfun$collect$1(RDD.scala:1049) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDD.withScope(RDD.scala:410) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDD.collect(RDD.scala:1048) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.PairRDDFunctions.$anonfun$countByKey$1(PairRDDFunctions.scala:367) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDD.withScope(RDD.scala:410) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.PairRDDFunctions.countByKey(PairRDDFunctions.scala:367) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.api.java.JavaPairRDD.countByKey(JavaPairRDD.scala:314) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.hudi.data.HoodieJavaPairRDD.countByKey(HoodieJavaPairRDD.java:109) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.buildProfile(BaseSparkCommitActionExecutor.java:195) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.execute(BaseSparkCommitActionExecutor.java:166) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.execute(BaseSparkCommitActionExecutor.java:83) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.table.action.commit.BaseWriteHelper.write(BaseWriteHelper.java:58) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
... 53 more
Caused by: org.apache.hudi.exception.HoodieException: org.apache.hudi.exception.HoodieException: Error occurs when executing map
at jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[?:?]
at jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:77) ~[?:?]
at jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) ~[?:?]
at java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:499) ~[?:?]
at java.lang.reflect.Constructor.newInstance(Constructor.java:480) ~[?:?]
at java.util.concurrent.ForkJoinTask.getThrowableException(ForkJoinTask.java:562) ~[?:?]
at java.util.concurrent.ForkJoinTask.reportException(ForkJoinTask.java:591) ~[?:?]
at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:689) ~[?:?]
at java.util.stream.ReduceOps$ReduceOp.evaluateParallel(ReduceOps.java:927) ~[?:?]
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233) ~[?:?]
at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682) ~[?:?]
at org.apache.hudi.common.engine.HoodieLocalEngineContext.map(HoodieLocalEngineContext.java:90) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieBackedTableMetadata.getAllRecordsByKeys(HoodieBackedTableMetadata.java:322) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.BaseTableMetadata.readRecordIndex(BaseTableMetadata.java:299) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.index.SparkMetadataTableRecordIndex$RecordIndexFileGroupLookupFunction.call(SparkMetadataTableRecordIndex.java:166) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.index.SparkMetadataTableRecordIndex$RecordIndexFileGroupLookupFunction.call(SparkMetadataTableRecordIndex.java:153) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsToPair$1(JavaRDDLike.scala:186) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:858) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:858) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.rdd.RDD.iterator(RDD.scala:331) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:104) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:54) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.scheduler.Task.run(Task.scala:141) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:620) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64) ~[spark-common-utils_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61) ~[spark-common-utils_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:623) ~[spark-core_2.12-3.5.2.jar:3.5.2]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
at java.lang.Thread.run(Thread.java:833) ~[?:?]
Caused by: org.apache.hudi.exception.HoodieException: Error occurs when executing map
at org.apache.hudi.common.function.FunctionWrapper.lambda$throwingMapWrapper$0(FunctionWrapper.java:40) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) ~[?:?]
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625) ~[?:?]
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[?:?]
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[?:?]
at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:960) ~[?:?]
at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:934) ~[?:?]
at java.util.stream.AbstractTask.compute(AbstractTask.java:327) ~[?:?]
at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:754) ~[?:?]
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:373) ~[?:?]
at java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1182) ~[?:?]
at java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1655) ~[?:?]
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1622) ~[?:?]
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:165) ~[?:?]
Caused by: org.apache.hudi.exception.HoodieException: Exception when reading log file
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.scanInternalV2(AbstractHoodieLogRecordScanner.java:604) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.table.log.AbstractHoodieLogRecordScanner.scanInternal(AbstractHoodieLogRecordScanner.java:248) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner.scanByFullKeys(HoodieMergedLogRecordScanner.java:163) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieMetadataLogRecordReader.getAllRecordsByKeys(HoodieMetadataLogRecordReader.java:127) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
at org.apache.hudi.metadata.HoodieBackedTableMetadata.readAllLogRecords(HoodieBackedTableMetadata.java:477) ~[hudi-spark3.5-bundle_2.12-1.0.1.jar:1.0.1]
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "robbik (via GitHub)" <gi...@apache.org>.
robbik commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2710086485
Hi @rangareddy
I am afraid it depends on the data itself. I tried the same code to process different dataset (from initial / empty S3 bucket, not using the existing record index files) and couldn't reproduce the issue. But if you use existing record index files with same incremental dataset, it is reproducible.
The record index files are quite big (~ 1GB) and the incremental dataset is around 100 thousand rows. I can share the record index files but I can't share the incremental dataset since it is our production data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "robbik (via GitHub)" <gi...@apache.org>.
robbik commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2820320563
Thank you for the fixings, I will retest it and let you know the result
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "robbik (via GitHub)" <gi...@apache.org>.
robbik commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2735561591
Hi @rangareddy , sorry for late reply. I am trying to reproduce the issue and finally I was able to reproduce the issue using minimal data. Here is the code I use to reproduce the issue.
```
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.config.HoodieMetaserverConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.config.RecordMergeMode;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.config.HoodieCleanConfig;
import org.apache.hudi.config.HoodieCompactionConfig;
import org.apache.hudi.config.HoodieIndexConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.hive.HiveSyncConfig;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
import org.apache.hudi.sync.common.HoodieSyncConfig;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import java.io.Serial;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class Test3 implements Serializable {
@Serial
private static final long serialVersionUID = -9019298349538148232L;
private static final String STAGE_PRIMARY_KEY_COLUMN_NAME_0 = "OurBranchID";
private static final String STAGE_PRIMARY_KEY_COLUMN_NAME_1 = "AccountID";
private final SparkSession spark;
protected final String targetTableName;
protected final String targetPath;
protected final String[] targetPrimaryKeyColumns;
protected final String[] targetPartitionColumnAndTypes;
protected final String[] targetIndexColumns;
protected final int targetParallelism;
public Test3(SparkSession spark) {
this.spark = spark;
this.targetTableName = "loan_installment";
this.targetPath = "/tmp/test3";
this.targetPrimaryKeyColumns = new String[] {
STAGE_PRIMARY_KEY_COLUMN_NAME_0,
STAGE_PRIMARY_KEY_COLUMN_NAME_1
};
this.targetPartitionColumnAndTypes = new String[] { "OurBranchID:SIMPLE" };
this.targetIndexColumns = new String[0];
this.targetParallelism = 1;
}
protected String[] extractColumnNames(String... columnNameAndTypes) {
final String[] r = new String[columnNameAndTypes.length];
for (int i = 0; i < r.length; i++) {
final String v = columnNameAndTypes[i];
if (v.contains(":")) {
r[i] = v.split(":", 2)[0];
} else {
r[i] = v;
}
}
return r;
}
protected void writeTarget(Dataset<Row> data, boolean truncateFirst) {
final Map<String, String> opts = new HashMap<>();
opts.put(HoodieCleanConfig.CLEANER_POLICY.key(), "KEEP_LATEST_COMMITS");
opts.put(HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key(), "10");
opts.put(HoodieCompactionConfig.INLINE_COMPACT.key(), "true");
opts.put(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key(), "20");
opts.put(HoodieMetadataConfig.ENABLE_OPTIMIZED_LOG_BLOCKS_SCAN.key(), "true");
opts.put(HoodieSyncConfig.META_SYNC_ENABLED.key(), "false");
opts.put(HiveSyncConfig.HIVE_SYNC_ENABLED.key(), "false");
opts.put("hoodie.datasource.meta.sync.glue.partition_index_fields.enable", "false");
opts.put(HoodieTableConfig.TYPE.key(), HoodieTableType.MERGE_ON_READ.name());
opts.put("hoodie.datasource.write.table.type", HoodieTableType.MERGE_ON_READ.name());
opts.put(HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT.key(), "0");
opts.put(HoodieStorageConfig.PARQUET_COMPRESSION_CODEC_NAME.key(), "snappy");
opts.put(HoodieWriteConfig.RECORD_MERGE_MODE.key(), RecordMergeMode.COMMIT_TIME_ORDERING.name());
if (truncateFirst) {
opts.put("hoodie.datasource.write.operation", "insert_overwrite");
opts.put(HoodieWriteConfig.BULKINSERT_PARALLELISM_VALUE.key(), Integer.toString(targetParallelism));
} else {
opts.put("hoodie.datasource.write.operation", "upsert");
opts.put(HoodieWriteConfig.UPSERT_PARALLELISM_VALUE.key(), Integer.toString(targetParallelism));
}
opts.put(HoodieWriteConfig.INSERT_PARALLELISM_VALUE.key(), Integer.toString(targetParallelism));
opts.put(HoodieCleanConfig.CLEANER_PARALLELISM_VALUE.key(), Integer.toString(targetParallelism));
opts.put(HoodieMetadataConfig.ENABLE.key(), "true");
opts.put(HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key(), "true");
opts.put(HoodieMetadataConfig.ENABLE_METADATA_INDEX_PARTITION_STATS.key(), "true");
opts.put(HoodieMetadataConfig.ENABLE_METADATA_INDEX_BLOOM_FILTER.key(), "true");
opts.put(HoodieIndexConfig.INDEX_TYPE.key(), HoodieIndex.IndexType.RECORD_INDEX.name());
opts.put(HoodieIndexConfig.RECORD_INDEX_UPDATE_PARTITION_PATH_ENABLE.key(), "true");
opts.put(HoodieIndexConfig.RECORD_INDEX_USE_CACHING.key(), "true");
opts.put(HoodieMetadataConfig.RECORD_INDEX_ENABLE_PROP.key(), "true");
opts.put(HoodieMetadataConfig.RECORD_INDEX_MAX_FILE_GROUP_COUNT_PROP.key(), "500000000");
opts.put(HoodieMetadataConfig.RECORD_INDEX_MIN_FILE_GROUP_COUNT_PROP.key(), "1");
opts.put(HoodieIndexConfig.BLOOM_INDEX_UPDATE_PARTITION_PATH_ENABLE.key(), "true");
opts.put(HoodieIndexConfig.BLOOM_INDEX_USE_CACHING.key(), "true");
opts.put(HoodieIndexConfig.BLOOM_INDEX_USE_METADATA.key(), "true");
opts.put(HoodieMetaserverConfig.TABLE_NAME.key(), targetTableName);
opts.put(HoodieSyncConfig.META_SYNC_TABLE_NAME.key(), targetTableName);
opts.put(HoodieWriteConfig.TBL_NAME.key(), targetTableName);
opts.put(KeyGeneratorOptions.HIVE_STYLE_PARTITIONING_ENABLE.key(), "true");
opts.put(HoodieSyncConfig.META_SYNC_PARTITION_EXTRACTOR_CLASS.key(), "org.apache.hudi.hive.MultiPartKeysValueExtractor");
opts.put(HoodieSyncConfig.META_SYNC_PARTITION_FIELDS.key(), String.join(",", extractColumnNames(targetPartitionColumnAndTypes)));
opts.put(HoodieWriteConfig.KEYGENERATOR_CLASS_NAME.key(), "org.apache.hudi.keygen.CustomKeyGenerator");
opts.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), String.join(",", targetPrimaryKeyColumns));
opts.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(), String.join(",", targetPartitionColumnAndTypes));
opts.put("path", targetPath);
data.write()
.format("org.apache.hudi")
.options(opts)
.mode(truncateFirst ? SaveMode.Overwrite : SaveMode.Append)
.save();
}
public void fetchIncremental() throws Exception {
List<Row> rows = new ArrayList<>();
rows.add(RowFactory.create("0324", "0324NCA16778", "ACCOUNT NAME XYZ"));
rows.add(RowFactory.create("0324", "0324NCA16778A", "ACCOUNT NAME XYZ-A"));
final Dataset<Row> source = spark.createDataFrame(rows, DataTypes.createStructType(List.of(
DataTypes.createStructField("OurBranchID", DataTypes.StringType, true),
DataTypes.createStructField("AccountID", DataTypes.StringType, true),
DataTypes.createStructField("AccountName", DataTypes.StringType, true)
)));
writeTarget(source, false);
}
public void fetchAll() throws Exception {
List<Row> rows = new ArrayList<>();
for (int i = 1; i < 20_000; i += 2) {
rows.add(RowFactory.create("0324", "0324NCA" + String.format("%05d", i), "ACCOUNT NAME " + i));
}
rows.add(RowFactory.create("0324", "0324NH", "NH"));
rows.add(RowFactory.create("0324", "0324NT700001", "NT700001"));
rows.add(RowFactory.create("0324", "0324REG00002", "REG00002"));
final Dataset<Row> source = spark.createDataFrame(rows, DataTypes.createStructType(List.of(
DataTypes.createStructField("OurBranchID", DataTypes.StringType, true),
DataTypes.createStructField("AccountID", DataTypes.StringType, true),
DataTypes.createStructField("AccountName", DataTypes.StringType, true)
)));
writeTarget(source, true);
}
protected static SparkSession createSparkSession() {
final SparkConf sparkConfig = new SparkConf()
.loadFromSystemProperties(true)
.setAppName("test2");
sparkConfig.set("spark.ui.enabled", "true");
sparkConfig.set("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.hudi.catalog.HoodieCatalog");
sparkConfig.set("spark.sql.hive.convertMetastoreParquet", "false");
sparkConfig.set("spark.driver.maxResultSize", "256m");
sparkConfig.set("spark.memory.fraction", "0.6");
sparkConfig.set("spark.memory.storageFraction", "0.2");
sparkConfig.set("spark.rdd.compress", "true");
sparkConfig.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
sparkConfig.set("spark.kryo.registrator", "org.apache.spark.HoodieSparkKryoRegistrar");
sparkConfig.set("spark.kryoserializer.buffer.max", "256m");
sparkConfig.set("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension");
sparkConfig.set("spark.sql.parquet.datetimeRebaseModeInRead", "LEGACY");
sparkConfig.setMaster("local[6]");
final SparkContext sparkContext = new SparkContext(sparkConfig);
sparkContext.setLogLevel("ERROR");
return SparkSession.builder()
.config(sparkConfig)
.sparkContext(sparkContext)
.getOrCreate();
}
public static void main(String[] args) throws Exception {
SparkSession sparkSession = createSparkSession();
try {
new Test3(sparkSession).fetchAll();
} finally {
sparkSession.close();
}
sparkSession = createSparkSession();
try {
new Test3(sparkSession).fetchIncremental();
} finally {
sparkSession.close();
}
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "robbik (via GitHub)" <gi...@apache.org>.
robbik commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2785003500
@rangareddy any update on this issue?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] MOR Table upsert encountering java.lang.IllegalStateException: The current lookup key is less than the current position of the cursor, i.e., backward seekTo, which is not supported and should be avoided [hudi]
Posted by "rangareddy (via GitHub)" <gi...@apache.org>.
rangareddy commented on issue #12931:
URL: https://github.com/apache/hudi/issues/12931#issuecomment-2706977489
Hi @robbik
The issue mentioned above occurred while performing an upsert operation using a MOR table. I can see the exception from the test class. Could you please elaborate on when this issue occurred?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org