[HUDI-9125] Pass compaction/merge related props to HoodieBaseFileGroupRecordBuffer (#12925)

* feat: pass compaction/merge related props to HoodieBaseFileGroupRecordBuffer

1. pass compaction/merge related props to HoodieBaseFileGroupRecordBuffer

Signed-off-by: TheR1sing3un <[email protected]>

* fix: resolve multiple precombine-related configuration conflicts

1. resolve multiple precombine-related configuration conflicts
2. assume that precombine is based on table config

Signed-off-by: TheR1sing3un <[email protected]>

* style: simplify lambda expression

1. simplify lambda expression

Signed-off-by: TheR1sing3un <[email protected]>

* fix: fix the `record_key` and `_hoodie_record_key` are not mapped when the record is created in SparkDatasetTestUtils

1. fix the `record_key` and `_hoodie_record_key` are not mapped when the record is created in SparkDatasetTestUtils

Signed-off-by: TheR1sing3un <[email protected]>

* rerun

* feat: Remove the default value for PAYLOAD_ORDERING_FIELD_PROP_KEY to avoid taking the default value from props as a valid configuration

1. Remove the default value for PAYLOAD_ORDERING_FIELD_PROP_KEY to avoid taking the default value from props as a valid configuration优化排序字段获取逻辑，移除默认值配置。

Signed-off-by: TheR1sing3un <[email protected]>

* Update hudi-common/src/main/java/org/apache/hudi/common/util/collection/ExternalSpillableMap.java

---------

Signed-off-by: TheR1sing3un <[email protected]>
Co-authored-by: Y Ethan Guo <[email protected]>

Author: TheR1sing3un

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodiePayloadConfig.java

@@ -60,7 +60,7 @@ public class HoodiePayloadConfig extends HoodieConfig {
   @Deprecated
   public static final ConfigProperty<String> ORDERING_FIELD = ConfigProperty
       .key(PAYLOAD_ORDERING_FIELD_PROP_KEY)
-      .defaultValue("ts")
+      .noDefaultValue()
       .markAdvanced()
       .withDocumentation("Table column/field name to order records that have the same key, before "
           + "merging and writing to storage.");

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/HoodieSparkFileGroupReaderBasedMergeHandle.java

@@ -21,6 +21,7 @@
 
 import org.apache.hudi.AvroConversionUtils;
 import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.config.TypedProperties;
 import org.apache.hudi.common.engine.HoodieReaderContext;
 import org.apache.hudi.common.engine.TaskContextSupplier;
 import org.apache.hudi.common.fs.FSUtils;
@@ -180,6 +181,9 @@ public void write() {
     if (!StringUtils.isNullOrEmpty(config.getInternalSchema())) {
       internalSchemaOption = SerDeHelper.fromJson(config.getInternalSchema());
     }
+    TypedProperties props = new TypedProperties();
+    hoodieTable.getMetaClient().getTableConfig().getProps().forEach(props::putIfAbsent);
+    config.getProps().forEach(props::putIfAbsent);
     // Initializes file group reader
     try (HoodieFileGroupReader<T> fileGroupReader = new HoodieFileGroupReader<>(
         readerContext,
@@ -191,7 +195,7 @@ public void write() {
         writeSchemaWithMetaFields,
         internalSchemaOption,
         hoodieTable.getMetaClient(),
-        hoodieTable.getMetaClient().getTableConfig().getProps(),
+        props,
         0,
         Long.MAX_VALUE,
         usePosition)) {

hudi-common/src/main/java/org/apache/hudi/common/util/collection/ExternalSpillableMap.java

@@ -102,6 +102,7 @@ public ExternalSpillableMap(long maxInMemorySizeInBytes, String baseFilePath, Si
     this.isCompressionEnabled = isCompressionEnabled;
     this.valueSerializer = valueSerializer;
     this.loggingContext = loggingContext;
+    LOG.debug("Initializing ExternalSpillableMap with baseFilePath = {}, maxInMemorySizeInBytes = {}, diskMapType = {}", maxInMemorySizeInBytes, baseFilePath, diskMapType);
   }
 
   private void initDiskBasedMap() {

hudi-spark-datasource/hudi-spark-common/src/test/java/org/apache/hudi/testutils/SparkDatasetTestUtils.java

@@ -141,17 +141,18 @@ public static Dataset<Row> getRandomRowsWithKeys(SQLContext sqlContext,
    */
   public static Row getRandomValue(String partitionPath, boolean isError, String commitTime) {
     // order commit time, seq no, record key, partition path, file name
+    String recordKey = UUID.randomUUID().toString();
     Object[] values = new Object[9];
     values[0] = commitTime; //commit time
     if (!isError) {
       values[1] = ""; // commit seq no
     } else {
       values[1] = RANDOM.nextLong();
     }
-    values[2] = UUID.randomUUID().toString();
+    values[2] = recordKey;
     values[3] = partitionPath;
     values[4] = ""; // filename
-    values[5] = UUID.randomUUID().toString();
+    values[5] = recordKey;
     values[6] = partitionPath;
     values[7] = RANDOM.nextInt();
     if (!isError) {
@@ -180,7 +181,7 @@ public static Row getRandomValue(HoodieKey key, boolean isError, String commitTi
     values[2] = key.getRecordKey();
     values[3] = key.getPartitionPath();
     values[4] = ""; // filename
-    values[5] = UUID.randomUUID().toString();
+    values[5] = key.getRecordKey();
     values[6] = key.getPartitionPath();
     values[7] = RANDOM.nextInt();
     if (!isError) {