[SUPPORT] Hudi upsert takes more time than merging using spark sql

Spark version: 3.3

Issue:

1. I have 39GB parquet file on s3 which is ingested into Apache hudi. This is snappy compressed.
2. I have 147GB json file-s on s3 representing CDC data. This is CDC from mongo db.
3. each row in json file is ~ 5-6kb (kilo bytes)
4. -- When I try to merge them using spark SQL I see it completing in 9000 vcore seconds (This runs every day in our production. The SQL runs on 8-10 node EMR cluster ( 10 X M5.2x -8cores 32GB ram; ~8-10 executors are spinned on avg ). The same job when run with apache hudi takes 30000 (approx 22-30k vcore seconds)

I remember reading hudi is around LSM trees. And the way I understand COW of hudi is
1. Hudi gets a bunch of updates aimed at a parquet. Then suddenly copies the old parquet into a new one with the updates in it.
2. Essentially this mechanism, in my view, should be much much faster than Spark SQL. HOwever its not the case.

One suggestion i got is to partition the data. But this requires full pipeline change. Any ways of getting this less than spark SQL vcore secs?

spark shell command
```
spark-shell --driver-memory 1g --executor-memory 3g --executor-cores 1 --driver-cores 1 --conf spark.executor.heartbeatInterval=600s --conf spark.network.timeout=5000s --conf yarn.resourcemanager.nodemanagers.heartbeat-interval-max-ms=60000 --conf yarn.resourcemanager.nodemanagers.heartbeat-interval-min-ms=60000 --conf yarn.resourcemanager.nodemanagers.heartbeat-interval-ms=60000 --conf yarn.app.mapreduce.am.scheduler.heartbeat.interval-ms=60000 --conf yarn.app.mapreduce.am.hard-kill-timeout-ms=600000 --conf yarn.nodemanager.health-checker.timeout-ms=72000000 --conf yarn.nodemanager.health-checker.interval-ms=36000000 --conf yarn.resourcemanager.application-timeouts.monitor.interval-ms=180000 --conf "spark.serializer=org.apache.spark.serializer.KryoSerializer" --conf "spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension" --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog" --conf "spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar" --conf spark.sql.adaptive.enabled=true --conf spark.sql.adaptive.coalescePartitions.enabled=true --conf spark.sql.adaptive.coalescePartitions.minPartitionNum=1 --conf spark.sql.adaptive.advisoryPartitionSizeInBytes="128MB" --conf spark.sql.legacy.parquet.int96RebaseModeInRead=CORRECTED --conf spark.sql.legacy.parquet.int96RebaseModeInWrite=CORRECTED --conf spark.sql.legacy.parquet.datetimeRebaseModeInRead=CORRECTED --conf spark.sql.legacy.parquet.datetimeRebaseModeInWrite=CORRECTED --conf spark.yarn.maxAppAttempts=1 --conf spark.yarn.maxAppAttempts=1 --conf spark.yarn.submit.waitAppCompletion=false --conf "spark.executor.extraJavaOptions=-Dlog4j.debug=false -XX:-PrintGC -XX:-PrintGCDetails -Dconfig.appname=ravic -Dconfig=s3://bucket/configs/ravic.json -Dconfig.properties=s3://bucket/configs/properties.json" --conf "spark.driver.extraJavaOptions=-Dlog4j.debug=false -XX:-PrintGC -XX:-PrintGCDetails -Dconfig.appname=ravic -Dconfig=s3://bucket/configs/ravic.json -Dconfig.properties=s3://bucket/configs/properties.json" --files /home/hadoop/jars/log4j2.properties --conf "spark.yarn.appMasterEnv.configs=s3://bucket/configs/ravic.json" --conf "spark.hadoop.parquet.avro.write-old-list-structure=false" --conf "spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=2" --conf spark.memory.fraction=0.8 --name ravic --packages org.apache.hudi:hudi-spark3.3-bundle_2.12:0.13.1 --jars /home/hadoop/jars2/spark-1.0-SNAPSHOT.jar

```

hudi code to merge

```
import org.apache.commons.lang3.ClassUtils.getCanonicalName
import org.apache.hudi.bootstrap.SparkParquetBootstrapDataProvider
import org.apache.hudi.{DataSourceWriteOptions, QuickstartUtils}
import org.apache.hudi.common.model.{HoodieAvroPayload, HoodieFileFormat, WriteOperationType}
import org.apache.hudi.common.table.HoodieTableConfig
import org.apache.hudi.config.{HoodieBootstrapConfig, HoodieWriteConfig}
import org.apache.hudi.keygen.constant.KeyGeneratorOptions

import java.util
import org.apache.hudi.config.HoodieWriteConfig.TBL_NAME
import org.apache.hudi.keygen.{NonpartitionedKeyGenerator, SimpleKeyGenerator}
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.functions.{col, hash, lit}
import org.apache.hudi.QuickstartUtils._

val sess = Application.spark();
/* get snapshot df; cdc df */
val snapshotDf = sess.read.parquet("s3://bucket/snapshots-test/ge11/_bid_9223370348396273945__mongoToParq/")
val cdcSchema1 = SparkUtils.getSchema("s3://bucket/schemas/ge11.json")
val cdcDf = sess.read.schema(snapshotDf.schema).json("s3://bucket/inputs-test/ge11/23-07-26/*")

cdcDf.createOrReplaceTempView("cdc")
val _cdcDf =sess.sql("select * from cdc where _id.oid is not null and _id.oid !='' ")
_cdcDf.createOrReplaceTempView("_cdc");
_cdcDf.write.format("hudi")
.options(getQuickstartWriteConfigs)
.option(HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key(), "cdc_pk")
.option(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "_id.oid")
.option(DataSourceWriteOptions.OPERATION.key(), WriteOperationType.UPSERT.name() )
.option(TBL_NAME.key(), "GE11")
.mode(SaveMode.Append)
.save("s3://buket/snapshots-hudi/ge11/snapshot");
```

SQL used to merge the files

```
with cdcTable as
(
select cdc_pk,
cdc_oid,
@{getCols('s3://bucket/schemas/[dbname]-[collection].json' ) }
from input where cdc_pk in (select max(cdc_pk) from input group by cdc_oid)
),
snapshotTable as
(
select cdc_pk,
cdc_oid,
@{getCols('s3://bucket/schemas/[dbname]-[collection].json') }
from snapshot where _id.oid not in (select cdc_oid from input where cdc_oid is not null)
),
resultTable as
(
select * from snapshotTable
union all
select * from cdcTable
)
select * from resultTable
```

`getCols` --> goes to the schema file and lists all columns one by one.

Start with the provided spark-shell command, Hudi 0.13.1 write block, and SQL merge query, then reproduce the comparison on Spark 3.3 using the stated snapshot and CDC sizes. Measure both jobs and inspect the Hudi upsert configuration and workload characteristics; done would be an evidenced explanation and an actionable change that addresses the reported performance gap.

Written by the indexing model from the issue text.