feat(core): introduce pluggable RecordMerger trait for file-group merging

crates/core/src/config/read.rs

@@ -28,6 +28,7 @@ use crate::config::Result;
 use crate::config::error::ConfigError;
 use crate::config::error::ConfigError::{InvalidValue, NotFound, ParseBool, ParseInt};
 use crate::config::{ConfigParser, HudiConfigValue};
+use crate::merge::DEFAULT_RECORD_MERGER_IMPL;
 
 /// Config value for [`HudiReadConfig::QueryType`]. Canonical strings are
 /// `snapshot` and `incremental`; [`FromStr`] accepts case-insensitive forms.
@@ -104,6 +105,9 @@ pub enum HudiReadConfig {
 
     /// Maximum number of file-slice streams to poll concurrently within one scan partition.
     FileSliceReadConcurrency,
+
+    /// Record merger implementation used for file-slice merging.
+    RecordMergerImpl,
 }
 
 impl HudiReadConfig {
@@ -120,6 +124,7 @@ impl HudiReadConfig {
             Self::UseReadOptimizedMode => "hoodie.read.use.read_optimized.mode",
             Self::StreamBatchSize => "hoodie.read.stream.batch_size",
             Self::FileSliceReadConcurrency => "hoodie.read.file.slice.read.concurrency",
+            Self::RecordMergerImpl => "hoodie.read.record.merger.impl",
         }
     }
 }
@@ -148,6 +153,9 @@ impl ConfigParser for HudiReadConfig {
             HudiReadConfig::UseReadOptimizedMode => Some(HudiConfigValue::Boolean(false)),
             HudiReadConfig::StreamBatchSize => Some(HudiConfigValue::UInteger(1024usize)),
             HudiReadConfig::FileSliceReadConcurrency => Some(HudiConfigValue::UInteger(4usize)),
+            HudiReadConfig::RecordMergerImpl => Some(HudiConfigValue::String(
+                DEFAULT_RECORD_MERGER_IMPL.to_string(),
+            )),
             _ => None,
         }
     }
@@ -197,6 +205,7 @@ impl ConfigParser for HudiReadConfig {
                     Ok(parsed)
                 })
                 .map(HudiConfigValue::UInteger),
+            Self::RecordMergerImpl => get_result.map(|v| HudiConfigValue::String(v.to_string())),
         }
     }
 }
@@ -206,7 +215,8 @@ mod tests {
     use super::*;
     use crate::config::read::HudiReadConfig::{
         AsOfTimestamp, EndTimestamp, FileSliceReadConcurrency, InputPartitions,
-        QueryType as QueryTypeKey, StartTimestamp, StreamBatchSize, UseReadOptimizedMode,
+        QueryType as QueryTypeKey, RecordMergerImpl, StartTimestamp, StreamBatchSize,
+        UseReadOptimizedMode,
     };
 
     #[test]
@@ -226,6 +236,10 @@ mod tests {
                 FileSliceReadConcurrency.as_ref().to_string(),
                 "8".to_string(),
             ),
+            (
+                RecordMergerImpl.as_ref().to_string(),
+                DEFAULT_RECORD_MERGER_IMPL.to_string(),
+            ),
         ]);
         let actual: String = QueryTypeKey.parse_value(&options).unwrap().into();
         assert_eq!(actual, "incremental");
@@ -246,6 +260,8 @@ mod tests {
             .unwrap()
             .into();
         assert_eq!(actual, 8);
+        let actual: String = RecordMergerImpl.parse_value(&options).unwrap().into();
+        assert_eq!(actual, DEFAULT_RECORD_MERGER_IMPL);
     }
 
     #[test]
@@ -292,6 +308,8 @@ mod tests {
             .parse_value_or_default(&options)
             .into();
         assert_eq!(actual, 4);
+        let actual: String = RecordMergerImpl.parse_value_or_default(&options).into();
+        assert_eq!(actual, DEFAULT_RECORD_MERGER_IMPL);
 
         let zero = HashMap::from([(
             FileSliceReadConcurrency.as_ref().to_string(),
@@ -342,5 +360,9 @@ mod tests {
             format!("{}", HudiReadConfig::QueryType),
             "hoodie.read.query.type"
         );
+        assert_eq!(
+            format!("{}", HudiReadConfig::RecordMergerImpl),
+            "hoodie.read.record.merger.impl"
+        );
     }
 }

crates/core/src/file_group/reader.rs

@@ -32,7 +32,7 @@ use crate::file_group::file_slice::FileSlice;
 use crate::file_group::log_file::scanner::{LogFileScanner, ScanResult};
 use crate::file_group::record_batches::RecordBatches;
 use crate::hfile::{HFileReader, HFileRecord};
-use crate::merge::record_merger::RecordMerger;
+use crate::merge::create_record_merger;
 use crate::metadata::merger::FilesPartitionMerger;
 use crate::metadata::meta_field::MetaField;
 use crate::metadata::table_record::FilesPartitionRecord;
@@ -271,8 +271,8 @@ impl FileGroupReader {
             all_batches.push_data_batch(base_batch);
             all_batches.extend(log_batches);
 
-            let merger = RecordMerger::new(schema.clone(), self.hudi_configs.clone());
-            merger.merge_record_batches(all_batches)?
+            let merger = create_record_merger(schema.clone(), self.hudi_configs.clone())?;
+            merger.merge(all_batches)?
         };
 
         apply_eager_options(&options, merged)