Upsert operation not working and job is running longer while using "Record level index" in Apache Hudi 0.14 in EMR 6.15

**Describe the problem you faced**

I am facing issues for upsert operation in hudi 0.14 RLI in EMR 6.15 Spark 3.4.1 using "Record level Index".
i see insert mode working as expected but upsert operation is not working with existing hudi table(lower version) and having several issues. In short, I am getting error for hoodieException config conflict(key current value existing value)
Hoodie.database.name : database name
I am giving the same database name(hudi db) where data is stored as target and same db providing in source data ( hudi config) and trying to upsert but throwing this hoodie.database.name issues.
Can you please have a look and help me to fix this issues.

**To Reproduce**
Steps to reproduce the behavior (For me):
1. Spin up new EMR cluster (emr-6.15.0 with spark 3.4.1
2. placed the jar **hudi-spark3.4-bundle_2.12-0.14.0.jar** and spark-avro_2.13-3.5.0.jar on S3 jar location
3. trying to run the exact script which is running on emr-6.5.0 with Spark 3.1.2 and hudi-spark3.1.2-bundle_2.12-0.10.1.jar

**Expected behavior**
My expectations would be to have the same script run successfully and insert and update data into Hudi table.

**Environment Description**
Hudi version : 0.14.0
Spark version : 3.4.1
Hive version : 3.1.3
Hadoop version : 3.3.6
Storage (HDFS/S3/GCS..) : s3
Running on Docker? (yes/no) : no

**Additional context**
config I am using for upsert operation:
**Upsert operation config:**

hudi_operation = "upsert"
hudi_write_mode = "append"
hudi_config["className"] = "org.apache.hudi"
hudi_config["hoodie.table.keygenerator.class"] = "org.apache.hudi.keygen.ComplexKeyGenerator"
hudi_config["hoodie.table.name"] = tgt_tbl
hudi_config["hoodie.database.name"] = tgt_db
hudi_config["hoodie.datasource.write.recordkey.field"] = "id"
hudi_config["hoodie.datasource.write.operation"] = hudi_operation
hudi_config["hoodie.datasource.hive_sync.table"] = tgt_tbl
hudi_config["hoodie.datasource.hive_sync.database"] = tgt_db
hudi_config["hoodie.index.type"] = "RECORD_INDEX"
hudi_config["hoodie.datasource.write.table.name"] = tgt_tbl
hudi_config["hoodie.datasource.write.precombine.field"] = "eff_fm_cent_tz"
hudi_config["hoodie.metadata.record.index.enable"] = "true"

**below is code snippet :-**

hudi_config = {
"className":"org.apache.hudi",
"hoodie.table.name": tgt_tbl,
"hoodie.datasource.write.recordkey.field": "id",
"hoodie.datasource.write.precombine.field": "eff_fm_cent_tz",
"hoodie.datasource.write.operation": "upsert",
"hoodie.datasource.write.keygenerator.class": "org.apache.hudi.keygen.ComplexKeyGenerator",
"hoodie.datasource.write.partitionpath.field": "year,month",
"hoodie.datasource.hive_sync.support_timestamp": "true",
"hoodie.datasource.hive_sync.enable": "true",
"hoodie.datasource.hive_sync.assume_date_partitioning": "false",
"hoodie.datasource.hive_sync.table": tgt_tbl,
"hoodie.datasource.hive_sync.use_jdbc": "false",
"hoodie.datasource.hive_sync.mode": "hms",
"hoodie.datasource.hive_sync.partition_extractor_class": "org.apache.hudi.hive.MultiPartKeysValueExtractor",
"hoodie.datasource.write.hive_style_partitioning": "true",
"hoodie.upsert.shuffle.parallelism": hudi_upsert_parallelism,
"hoodie.delete.shuffle.parallelism": hudi_delete_parallelism,
"hoodie.bulkinsert.sort.mode": "GLOBAL_SORT",
"hoodie.metadata.enable": "true",
"hoodie.metadata.record.index.enable": "true",
"hoodie.index.type": "RECORD_INDEX",
"hoodie.metadata.index.column.stats.column.list": "res_sys_id,pnr_rec_loc_id,pnr_cre_dt",
"hoodie.enable.data.skipping": "true"
}
hudi_delete_config = {
"hoodie.datasource.write.payload.class":
"org.apache.hudi.common.model.EmptyHoodieRecordPayload"
}
# Initialize local s3
if (files_exist == True):
print("upsert started")
hudi_operation = "upsert"
hudi_write_mode = "append"
hudi_config["className"] = "org.apache.hudi"
hudi_config["hoodie.table.keygenerator.class"] = "org.apache.hudi.keygen.ComplexKeyGenerator"
hudi_config["hoodie.database.name"] = tgt_db
hudi_config["hoodie.table.name"] = tgt_tbl
hudi_config["hoodie.datasource.write.recordkey.field"] = "id"
hudi_config["hoodie.datasource.write.operation"] = hudi_operation
hudi_config["hoodie.datasource.hive_sync.table"] = tgt_tbl
hudi_config["hoodie.datasource.hive_sync.database"] = tgt_db
hudi_config["hoodie.index.type"] = "RECORD_INDEX"
hudi_config["hoodie.metadata.enable"] = "true"
hudi_config["hoodie.datasource.write.table.name"] = tgt_tbl
hudi_config["hoodie.datasource.write.precombine.field"] = "eff_fm_cent_tz"
hudi_config["hoodie.metadata.record.index.enable"] = "true"
print("Upserting records into " + tgt_tbl + " Hudi table")
res_pnr_concat.write.format("org.apache.hudi").options(**hudi_config).mode("Append").save(hudi_tbl_path + tgt_tbl)
print("Successfully process records for " + tgt_tbl + " Hudi table")
print("res_pnr hudi loading for upsert ended ----",datetime.now())
if len(target_records) > 0:
print("delete logic started ----",datetime.now())
res_pnr_concat.createOrReplaceTempView("res_pnr_concat")
res_pnr_del_df=spark.sql("select distinct A.* from res_pnr_cached A where not exists(select 1 from res_pnr_concat B where A.RES_SYS_ID = B.RES_SYS_ID AND A.PNR_REC_LOC_ID = B.PNR_REC_LOC_ID AND A.PNR_CRE_DT = B.PNR_CRE_DT AND A.EFF_FM_CENT_TZ = B.eff_fm_cent_tz and A.year = B.year and A.month = B.month)").drop("src_hoodie_record_key").drop("_hoodie_commit_seqno").drop("_hoodie_commit_time").drop("_hoodie_file_name").drop("_hoodie_partition_path").drop("_hoodie_record_key")
if len(res_pnr_del_df.take(1)) > 0:
common_config = {**hudi_config, **hudi_delete_config}
spark.sql("uncache table if exists res_pnr_src_df")
print("Deleting records from RES_PNR Hudi table")
res_pnr_del_df.write.format("org.apache.hudi").options(**common_config).mode("append").save(hudi_tbl_path + tgt_tbl)
else:
print("Delete eligible records are NOT identified")
print("delete logic Ended ----",datetime.now())
else:
pass
print("res_pnr hudi upsert logic ended ----",datetime.now())
else:
hudi_operation = "bulk_insert"
hudi_write_mode = "overwrite"
hudi_config["className"] = "org.apache.hudi"
hudi_config["hoodie.table.name"] = tgt_tbl
hudi_config["hoodie.database.name"] = tgt_db
hudi_config["hoodie.datasource.write.operation"] = hudi_operation
hudi_config["hoodie.datasource.hive_sync.table"] = tgt_tbl
hudi_config["hoodie.datasource.write.recordkey.field"] = "id"
hudi_config["hoodie.datasource.hive_sync.database"] = tgt_db
hudi_config["hoodie.index.type"] = "RECORD_INDEX"
hudi_config["hoodie.metadata.enable"] = "true"
hudi_config["hoodie.datasource.write.table.name"] = tgt_tbl
hudi_config["hoodie.datasource.write.precombine.field"] = "eff_fm_cent_tz"
hudi_config["hoodie.metadata.record.index.enable"] = "true"
print("bulk insert " + tgt_tbl + " Hudi table")
res_pnr_concat.write.format("org.apache.hudi").options(**hudi_config).mode(hudi_write_mode).save(hudi_tbl_path + tgt_tbl)
print("Successfully bulk upserted records into " + tgt_tbl + " Hudi table")
print("res_pnr hudi loading for upsert ended ----",datetime.now())

**Upsert operation job is throwing an error:**

An error was encountered:
An error occurred while calling o2149.save.
: org.apache.hudi.exception.HoodieException: Config conflict(key current value existing value):
hoodie.database.name: datalake_dev1_entp_cds
at org.apache.hudi.HoodieWriterUtils$.validateTableConfig(HoodieWriterUtils.scala:211)
at org.apache.hudi.HoodieSparkSqlWriter$.writeInternal(HoodieSparkSqlWriter.scala:177)
at org.apache.hudi.HoodieSparkSqlWriter$.write(HoodieSparkSqlWriter.scala:132)
at org.apache.hudi.DefaultSource.createRelation(DefaultSource.scala:150)
at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:47)
: org.apache.hudi.exception.HoodieException: Config conflict(key current value existing value):
hoodie.database.name: datalake_dev1_entp_cds
at org.apache.hudi.HoodieWriterUtils$.validateTableConfig(HoodieWriterUtils.scala:211)
at org.apache.hudi.HoodieSparkSqlWriter$.writeInternal(HoodieSparkSqlWriter.scala:177)
at org.apache.hudi.HoodieSparkSqlWriter$.write(HoodieSparkSqlWriter.scala:132)
at org.apache.hudi.DefaultSource.createRelation(DefaultSource.scala:150)

Start with the reported validation entry point at HoodieWriterUtils.scala:211 and follow the call from HoodieSparkSqlWriter.scala. Reproduce the upsert against the existing Hudi table using Hudi 0.14.0, Spark 3.4.1, and EMR 6.15 with the supplied configuration. Done means the cause of the hoodie.database.name conflict and the longer-running upsert behavior are established, with a verified outcome for the existing table.

Written by the indexing model from the issue text.