diff --git a/crates/core/src/config/read.rs b/crates/core/src/config/read.rs index 10b0e5b2d..68660705e 100644 --- a/crates/core/src/config/read.rs +++ b/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" + ); } } diff --git a/crates/core/src/file_group/reader.rs b/crates/core/src/file_group/reader.rs index df691160b..08e85d583 100644 --- a/crates/core/src/file_group/reader.rs +++ b/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) diff --git a/crates/core/src/merge/conformance_tests.rs b/crates/core/src/merge/conformance_tests.rs new file mode 100644 index 000000000..60b31725c --- /dev/null +++ b/crates/core/src/merge/conformance_tests.rs @@ -0,0 +1,352 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::config::HudiConfigs; +use crate::config::table::HudiTableConfig::{ + OrderingFields, PopulatesMetaFields, RecordMergeStrategy, +}; +use crate::file_group::record_batches::RecordBatches; +use crate::merge::RecordMerger; +use crate::merge::record_batch_merger::RecordBatchMerger; +use crate::metadata::meta_field::MetaField; +use arrow_array::{Array, Int32Array, RecordBatch, StringArray}; +use arrow_schema::{DataType, Field, Schema, SchemaRef}; +use std::sync::Arc; + +#[derive(Debug, Eq, PartialEq)] +struct ConformanceRow { + record_key: String, + commit_time: String, + commit_seqno: String, + ts: Option, + value: i32, +} + +type DataRow<'a> = (&'a str, &'a str, &'a str, Option, i32); + +fn create_schema() -> SchemaRef { + Arc::new(Schema::new(vec![ + Field::new(MetaField::CommitTime.as_ref(), DataType::Utf8, false), + Field::new(MetaField::CommitSeqno.as_ref(), DataType::Utf8, false), + Field::new(MetaField::RecordKey.as_ref(), DataType::Utf8, false), + Field::new("ts", DataType::Int32, true), + Field::new("value", DataType::Int32, false), + ])) +} + +fn append_only_configs() -> Arc { + Arc::new(HudiConfigs::new([ + (RecordMergeStrategy, "APPEND_ONLY"), + (PopulatesMetaFields, "false"), + ])) +} + +fn overwrite_with_latest_configs() -> Arc { + Arc::new(HudiConfigs::new([ + (RecordMergeStrategy, "OVERWRITE_WITH_LATEST"), + (PopulatesMetaFields, "true"), + (OrderingFields, "ts"), + ])) +} + +fn data_batch(schema: SchemaRef, rows: &[DataRow<'_>]) -> RecordBatch { + let commit_times: Vec<&str> = rows.iter().map(|row| row.0).collect(); + let commit_seqnos: Vec<&str> = rows.iter().map(|row| row.1).collect(); + let record_keys: Vec<&str> = rows.iter().map(|row| row.2).collect(); + let ordering_values: Vec> = rows.iter().map(|row| row.3).collect(); + let values: Vec = rows.iter().map(|row| row.4).collect(); + + RecordBatch::try_new( + schema, + vec![ + Arc::new(StringArray::from(commit_times)), + Arc::new(StringArray::from(commit_seqnos)), + Arc::new(StringArray::from(record_keys)), + Arc::new(Int32Array::from(ordering_values)), + Arc::new(Int32Array::from(values)), + ], + ) + .unwrap() +} + +fn delete_batch(rows: &[(&str, i32)]) -> RecordBatch { + let schema = Arc::new(Schema::new(vec![ + Field::new("recordKey", DataType::Utf8, false), + Field::new("partitionPath", DataType::Utf8, false), + Field::new("orderingVal", DataType::Int32, false), + ])); + let record_keys: Vec<&str> = rows.iter().map(|row| row.0).collect(); + let partition_paths: Vec<&str> = rows.iter().map(|_| "partition").collect(); + let ordering_values: Vec = rows.iter().map(|row| row.1).collect(); + + RecordBatch::try_new( + schema, + vec![ + Arc::new(StringArray::from(record_keys)), + Arc::new(StringArray::from(partition_paths)), + Arc::new(Int32Array::from(ordering_values)), + ], + ) + .unwrap() +} + +fn normalize_output(batch: &RecordBatch) -> Vec { + let schema = batch.schema(); + let commit_time_idx = schema.index_of(MetaField::CommitTime.as_ref()).unwrap(); + let commit_seqno_idx = schema.index_of(MetaField::CommitSeqno.as_ref()).unwrap(); + let record_key_idx = schema.index_of(MetaField::RecordKey.as_ref()).unwrap(); + let ts_idx = schema.index_of("ts").unwrap(); + let value_idx = schema.index_of("value").unwrap(); + + let commit_times = batch.column(commit_time_idx); + let commit_times = commit_times.as_any().downcast_ref::().unwrap(); + let commit_seqnos = batch.column(commit_seqno_idx); + let commit_seqnos = commit_seqnos + .as_any() + .downcast_ref::() + .unwrap(); + let record_keys = batch.column(record_key_idx); + let record_keys = record_keys.as_any().downcast_ref::().unwrap(); + let timestamps = batch.column(ts_idx); + let timestamps = timestamps.as_any().downcast_ref::().unwrap(); + let values = batch.column(value_idx); + let values = values.as_any().downcast_ref::().unwrap(); + + let mut rows: Vec = (0..batch.num_rows()) + .map(|idx| ConformanceRow { + record_key: record_keys.value(idx).to_string(), + commit_time: commit_times.value(idx).to_string(), + commit_seqno: commit_seqnos.value(idx).to_string(), + ts: if timestamps.is_null(idx) { + None + } else { + Some(timestamps.value(idx)) + }, + value: values.value(idx), + }) + .collect(); + rows.sort_by(|left, right| { + left.record_key + .cmp(&right.record_key) + .then_with(|| left.commit_time.cmp(&right.commit_time)) + .then_with(|| left.commit_seqno.cmp(&right.commit_seqno)) + .then_with(|| left.ts.cmp(&right.ts)) + .then_with(|| left.value.cmp(&right.value)) + }); + rows +} + +fn expected_rows(rows: &[DataRow<'_>]) -> Vec { + let schema = create_schema(); + let batch = data_batch(schema, rows); + normalize_output(&batch) +} + +fn run_conformance_suite(create_merger: F, schema: SchemaRef) +where + F: Fn(Arc) -> Arc, +{ + empty_input(create_merger(overwrite_with_latest_configs()).as_ref()); + append_only( + create_merger(append_only_configs()).as_ref(), + schema.clone(), + ); + overwrite_with_latest( + create_merger(overwrite_with_latest_configs()).as_ref(), + schema.clone(), + ); + deletes_only_remove_matching( + create_merger(overwrite_with_latest_configs()).as_ref(), + schema.clone(), + ); + mixed_insert_update_delete( + create_merger(overwrite_with_latest_configs()).as_ref(), + schema.clone(), + ); + null_ordering_values( + create_merger(overwrite_with_latest_configs()).as_ref(), + schema, + ); +} + +fn empty_input(merger: &dyn RecordMerger) { + let merged = merger.merge(RecordBatches::new()).unwrap(); + + assert_eq!(merged.num_rows(), 0); + assert_eq!(merged.schema(), merger.output_schema().clone()); +} + +fn append_only(merger: &dyn RecordMerger, schema: SchemaRef) { + let batch1 = data_batch( + schema.clone(), + &[ + ("20240101000000", "001", "k1", Some(1), 10), + ("20240101000000", "002", "k2", Some(2), 20), + ], + ); + let batch2 = data_batch( + schema, + &[ + ("20240102000000", "003", "k1", Some(3), 30), + ("20240102000000", "004", "k3", Some(4), 40), + ], + ); + + let merged = merger + .merge(RecordBatches::new_with_data_batches([batch1, batch2])) + .unwrap(); + + assert_eq!( + normalize_output(&merged), + expected_rows(&[ + ("20240101000000", "001", "k1", Some(1), 10), + ("20240102000000", "003", "k1", Some(3), 30), + ("20240101000000", "002", "k2", Some(2), 20), + ("20240102000000", "004", "k3", Some(4), 40), + ]) + ); +} + +fn overwrite_with_latest(merger: &dyn RecordMerger, schema: SchemaRef) { + let batch1 = data_batch( + schema.clone(), + &[ + ("20240101000000", "001", "k1", Some(1), 10), + ("20240101000000", "002", "k2", Some(5), 20), + ("20240101000000", "003", "k3", Some(3), 30), + ], + ); + let batch2 = data_batch( + schema, + &[ + ("20240102000000", "004", "k1", Some(4), 40), + ("20240102000000", "005", "k2", Some(2), 50), + ("20240102000000", "006", "k3", Some(3), 60), + ], + ); + + let merged = merger + .merge(RecordBatches::new_with_data_batches([batch1, batch2])) + .unwrap(); + + assert_eq!( + normalize_output(&merged), + expected_rows(&[ + ("20240102000000", "004", "k1", Some(4), 40), + ("20240101000000", "002", "k2", Some(5), 20), + ("20240102000000", "006", "k3", Some(3), 60), + ]) + ); +} + +fn deletes_only_remove_matching(merger: &dyn RecordMerger, schema: SchemaRef) { + let batch = data_batch( + schema, + &[ + ("20240101000000", "001", "k1", Some(1), 10), + ("20240101000000", "002", "k2", Some(2), 20), + ("20240101000000", "003", "k3", Some(3), 30), + ], + ); + let deletes = delete_batch(&[("k2", 5), ("missing", 5)]); + let mut batches = RecordBatches::new_with_data_batches([batch]); + batches.push_delete_batch(deletes, "20240102000000".to_string()); + + let merged = merger.merge(batches).unwrap(); + + assert_eq!( + normalize_output(&merged), + expected_rows(&[ + ("20240101000000", "001", "k1", Some(1), 10), + ("20240101000000", "003", "k3", Some(3), 30), + ]) + ); +} + +fn mixed_insert_update_delete(merger: &dyn RecordMerger, schema: SchemaRef) { + let batch1 = data_batch( + schema.clone(), + &[ + ("20240101000000", "001", "k1", Some(1), 10), + ("20240101000000", "002", "k2", Some(1), 20), + ("20240101000000", "003", "k3", Some(1), 30), + ], + ); + let batch2 = data_batch( + schema, + &[ + ("20240102000000", "004", "k1", Some(4), 40), + ("20240102000000", "005", "k4", Some(2), 50), + ], + ); + let deletes = delete_batch(&[("k1", 3), ("k2", 2)]); + let mut batches = RecordBatches::new_with_data_batches([batch1, batch2]); + batches.push_delete_batch(deletes, "20240103000000".to_string()); + + let merged = merger.merge(batches).unwrap(); + + assert_eq!( + normalize_output(&merged), + expected_rows(&[ + ("20240102000000", "004", "k1", Some(4), 40), + ("20240101000000", "003", "k3", Some(1), 30), + ("20240102000000", "005", "k4", Some(2), 50), + ]) + ); +} + +fn null_ordering_values(merger: &dyn RecordMerger, schema: SchemaRef) { + let batch1 = data_batch( + schema.clone(), + &[ + ("20240101000000", "001", "k1", Some(1), 10), + ("20240101000000", "002", "k2", None, 20), + ("20240101000000", "003", "k3", Some(3), 30), + ], + ); + let batch2 = data_batch( + schema, + &[ + ("20240102000000", "004", "k1", None, 40), + ("20240102000000", "005", "k2", Some(5), 50), + ], + ); + + let merged = merger + .merge(RecordBatches::new_with_data_batches([batch1, batch2])) + .unwrap(); + + assert_eq!( + normalize_output(&merged), + expected_rows(&[ + ("20240101000000", "001", "k1", Some(1), 10), + ("20240102000000", "005", "k2", Some(5), 50), + ("20240101000000", "003", "k3", Some(3), 30), + ]) + ); +} + +#[test] +fn record_batch_merger_satisfies_conformance() { + let schema = create_schema(); + let merger_schema = schema.clone(); + run_conformance_suite( + |configs| Arc::new(RecordBatchMerger::new(merger_schema.clone(), configs)), + schema, + ); +} diff --git a/crates/core/src/merge/mod.rs b/crates/core/src/merge/mod.rs index 6ca9f312d..d08c4cce2 100644 --- a/crates/core/src/merge/mod.rs +++ b/crates/core/src/merge/mod.rs @@ -16,15 +16,98 @@ * specific language governing permissions and limitations * under the License. */ +#[cfg(test)] +mod conformance_tests; mod ordering; -pub mod record_merger; +pub mod record_batch_merger; +use crate::Result; +use crate::config::HudiConfigs; use crate::config::error; use crate::config::error::ConfigError; use crate::config::error::ConfigError::InvalidValue; +use crate::config::error::Result as ConfigResult; +use crate::config::read::HudiReadConfig; +use crate::config::table::HudiTableConfig::{ + OrderingFields, PopulatesMetaFields, RecordMergeStrategy, +}; +use crate::error::CoreError; +use crate::file_group::record_batches::RecordBatches; +use crate::merge::record_batch_merger::RecordBatchMerger; +use arrow_array::RecordBatch; +use arrow_schema::SchemaRef; use std::str::FromStr; +use std::sync::Arc; use strum_macros::AsRefStr; +/// Merges materialized data and delete batches into a single output batch. +/// +/// Implementations are configured when constructed. Callers provide the union of +/// base-file and log-file data batches plus delete batches; the merger returns a +/// batch using [`Self::output_schema`]. Errors are returned for invalid +/// merge configuration, missing required fields, or Arrow compute failures. +pub trait RecordMerger: Send + Sync + std::fmt::Debug { + /// Merges the provided record batches into one output batch. + /// + /// The input must contain batches compatible with this merger's output + /// schema and configured Hudi merge strategy. Depending on that strategy, + /// implementations may apply deletes or deduplicate by record key using + /// ordering values. The returned batch uses [`Self::output_schema`]. + fn merge(&self, inputs: RecordBatches) -> Result; + + /// Returns the schema used by batches produced by this merger. + fn output_schema(&self) -> &SchemaRef; +} + +/// Default record merger implementation name. +pub const DEFAULT_RECORD_MERGER_IMPL: &str = "record_batch"; + +/// Creates a record merger from the configured merger implementation name. +pub fn create_record_merger( + schema: SchemaRef, + hudi_configs: Arc, +) -> Result> { + let impl_name: String = hudi_configs + .get_or_default(HudiReadConfig::RecordMergerImpl) + .into(); + let normalized_impl_name = impl_name.to_ascii_lowercase(); + match normalized_impl_name.as_str() { + DEFAULT_RECORD_MERGER_IMPL => Ok(Arc::new(RecordBatchMerger::new(schema, hudi_configs))), + _ => Err(CoreError::Config(ConfigError::InvalidValue(format!( + "unknown record_merger_impl: {impl_name}" + )))), + } +} + +/// Validates merge-related Hudi configs shared by all record merger implementations. +pub fn validate_configs(hudi_configs: &HudiConfigs) -> ConfigResult<()> { + let merge_strategy: String = hudi_configs.get_or_default(RecordMergeStrategy).into(); + let merge_strategy = RecordMergeStrategyValue::from_str(&merge_strategy)?; + + let populate_meta_fields: bool = hudi_configs.get_or_default(PopulatesMetaFields).into(); + if !populate_meta_fields && merge_strategy != RecordMergeStrategyValue::AppendOnly { + return Err(ConfigError::InvalidValue(format!( + "When {:?} is false, {:?} must be {:?}.", + PopulatesMetaFields, + RecordMergeStrategy, + RecordMergeStrategyValue::AppendOnly + ))); + } + + let precombine_field = hudi_configs.try_get(OrderingFields)?; + if precombine_field.is_none() && merge_strategy == RecordMergeStrategyValue::OverwriteWithLatest + { + return Err(ConfigError::InvalidValue(format!( + "When {:?} is {:?}, {:?} must be set.", + RecordMergeStrategy, + RecordMergeStrategyValue::OverwriteWithLatest, + OrderingFields + ))); + } + + Ok(()) +} + /// Config value for [crate::config::table::HudiTableConfig::RecordMergeStrategy]. #[derive(Clone, Debug, PartialEq, AsRefStr)] pub enum RecordMergeStrategyValue { @@ -45,3 +128,46 @@ impl FromStr for RecordMergeStrategyValue { } } } + +#[cfg(test)] +mod tests { + use super::*; + use arrow_array::Int32Array; + use arrow_schema::{DataType, Field, Schema}; + + fn create_test_schema() -> SchemaRef { + Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)])) + } + + fn create_test_batch(schema: SchemaRef) -> RecordBatch { + RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(vec![1, 2]))]).unwrap() + } + + #[test] + fn test_create_record_merger_default_merges_small_input() { + let schema = create_test_schema(); + let merger = create_record_merger(schema.clone(), Arc::new(HudiConfigs::empty())).unwrap(); + let batch = create_test_batch(schema.clone()); + + let merged = merger + .merge(RecordBatches::new_with_data_batches([batch])) + .unwrap(); + + assert_eq!(merger.output_schema(), &schema); + assert_eq!(merged.num_rows(), 2); + } + + #[test] + fn test_create_record_merger_unknown_impl_returns_config_error() { + let schema = create_test_schema(); + let configs = HudiConfigs::new([(HudiReadConfig::RecordMergerImpl, "bogus")]); + + let err = create_record_merger(schema, Arc::new(configs)).unwrap_err(); + + assert!(matches!( + err, + CoreError::Config(ConfigError::InvalidValue(msg)) + if msg == "unknown record_merger_impl: bogus" + )); + } +} diff --git a/crates/core/src/merge/record_merger.rs b/crates/core/src/merge/record_batch_merger.rs similarity index 81% rename from crates/core/src/merge/record_merger.rs rename to crates/core/src/merge/record_batch_merger.rs index 5e325069c..ae7822f01 100644 --- a/crates/core/src/merge/record_merger.rs +++ b/crates/core/src/merge/record_batch_merger.rs @@ -18,14 +18,10 @@ */ use crate::Result; use crate::config::HudiConfigs; -use crate::config::error::ConfigError; -use crate::config::error::Result as ConfigResult; -use crate::config::table::HudiTableConfig::{ - OrderingFields, PopulatesMetaFields, RecordMergeStrategy, -}; +use crate::config::table::HudiTableConfig::{OrderingFields, RecordMergeStrategy}; use crate::file_group::record_batches::RecordBatches; -use crate::merge::RecordMergeStrategyValue; use crate::merge::ordering::{MaxOrderingInfo, process_batch_for_max_orderings}; +use crate::merge::{RecordMergeStrategyValue, RecordMerger}; use crate::metadata::meta_field::MetaField; use crate::record::{ create_commit_time_ordering_converter, create_event_time_ordering_converter, @@ -42,43 +38,16 @@ use std::collections::HashMap; use std::str::FromStr; use std::sync::Arc; +/// Default record-batch merger using Hudi's current sort-and-deduplicate logic. #[derive(Debug, Clone)] -pub struct RecordMerger { +pub struct RecordBatchMerger { + /// Schema used by merged output batches. pub schema: SchemaRef, + /// Hudi configs that control merge strategy and ordering fields. pub hudi_configs: Arc, } -impl RecordMerger { - /// Validates the given [HudiConfigs] against the [RecordMergeStrategy]. - pub fn validate_configs(hudi_configs: &HudiConfigs) -> ConfigResult<()> { - let merge_strategy: String = hudi_configs.get_or_default(RecordMergeStrategy).into(); - let merge_strategy = RecordMergeStrategyValue::from_str(&merge_strategy)?; - - let populate_meta_fields: bool = hudi_configs.get_or_default(PopulatesMetaFields).into(); - if !populate_meta_fields && merge_strategy != RecordMergeStrategyValue::AppendOnly { - return Err(ConfigError::InvalidValue(format!( - "When {:?} is false, {:?} must be {:?}.", - PopulatesMetaFields, - RecordMergeStrategy, - RecordMergeStrategyValue::AppendOnly - ))); - } - - let precombine_field = hudi_configs.try_get(OrderingFields)?; - if precombine_field.is_none() - && merge_strategy == RecordMergeStrategyValue::OverwriteWithLatest - { - return Err(ConfigError::InvalidValue(format!( - "When {:?} is {:?}, {:?} must be set.", - RecordMergeStrategy, - RecordMergeStrategyValue::OverwriteWithLatest, - OrderingFields - ))); - } - - Ok(()) - } - +impl RecordBatchMerger { pub fn new(schema: SchemaRef, hudi_configs: Arc) -> Self { Self { schema, @@ -189,9 +158,21 @@ impl RecordMerger { } } +impl RecordMerger for RecordBatchMerger { + fn merge(&self, inputs: RecordBatches) -> Result { + self.merge_record_batches(inputs) + } + + fn output_schema(&self) -> &SchemaRef { + &self.schema + } +} + #[cfg(test)] mod tests { use super::*; + use crate::config::error::ConfigError; + use crate::config::table::HudiTableConfig::PopulatesMetaFields; use arrow_array::{Int32Array, StringArray}; use arrow_schema::{DataType, Field, Schema, SchemaRef}; @@ -216,21 +197,32 @@ mod tests { #[test] fn test_validate_configs() { - // Valid config with precombine field and meta fields - let configs = create_configs("OVERWRITE_WITH_LATEST", true, Some("ts")); - assert!(RecordMerger::validate_configs(&configs).is_ok()); - - // Valid append only config without meta fields - let configs = create_configs("APPEND_ONLY", false, None); - assert!(RecordMerger::validate_configs(&configs).is_ok()); - - // Invalid: Overwrite without precombine field - let configs = create_configs("OVERWRITE_WITH_LATEST", true, None); - assert!(RecordMerger::validate_configs(&configs).is_err()); - - // Invalid: No meta fields with overwrite strategy - let configs = create_configs("OVERWRITE_WITH_LATEST", false, Some("ts")); - assert!(RecordMerger::validate_configs(&configs).is_err()); + let cases = [ + ("OVERWRITE_WITH_LATEST", true, Some("ts"), None), + ("APPEND_ONLY", false, None, None), + ("OVERWRITE_WITH_LATEST", true, None, Some("OrderingFields")), + ( + "OVERWRITE_WITH_LATEST", + false, + Some("ts"), + Some("PopulatesMetaFields"), + ), + ]; + + for (strategy, populates_meta_fields, precombine, expected_error) in cases { + let configs = create_configs(strategy, populates_meta_fields, precombine); + match expected_error { + None => assert!(crate::merge::validate_configs(&configs).is_ok()), + Some(message_fragment) => { + let result = crate::merge::validate_configs(&configs); + assert!(matches!( + result, + Err(ConfigError::InvalidValue(message)) + if message.contains(message_fragment) + )); + } + } + } } fn create_schema(fields: Vec<(&str, DataType, bool)>) -> SchemaRef { @@ -300,7 +292,7 @@ mod tests { let schema = create_test_schema(false); let configs = create_configs("OVERWRITE_WITH_LATEST", true, Some("ts")); - let merger = RecordMerger::new(schema.clone(), Arc::new(configs)); + let merger = RecordBatchMerger::new(schema.clone(), Arc::new(configs)); // Test empty input let empty_result = merger.merge_record_batches(RecordBatches::new()).unwrap(); @@ -344,7 +336,7 @@ mod tests { .unwrap(); let configs = create_configs("APPEND_ONLY", false, None); - let merger = RecordMerger::new(schema.clone(), Arc::new(configs)); + let merger = RecordBatchMerger::new(schema.clone(), Arc::new(configs)); let merged = merger .merge_record_batches(RecordBatches::new_with_data_batches([batch1, batch2])) .unwrap(); @@ -395,7 +387,7 @@ mod tests { .unwrap(); let configs = create_configs("OVERWRITE_WITH_LATEST", true, Some("ts")); - let merger = RecordMerger::new(schema.clone(), Arc::new(configs)); + let merger = RecordBatchMerger::new(schema.clone(), Arc::new(configs)); let batches = RecordBatches::new_with_data_batches([batch1, batch2]); let merged = merger.merge_record_batches(batches).unwrap(); @@ -443,7 +435,7 @@ mod tests { .unwrap(); let configs = create_configs("OVERWRITE_WITH_LATEST", true, Some("ts")); - let merger = RecordMerger::new(schema.clone(), Arc::new(configs)); + let merger = RecordBatchMerger::new(schema.clone(), Arc::new(configs)); let batches = RecordBatches::new_with_data_batches([batch1, batch2]); let merged = merger.merge_record_batches(batches).unwrap(); @@ -459,4 +451,44 @@ mod tests { ] ); } + + #[test] + fn test_record_merger_trait_merge_matches_direct_merge() { + let schema = create_test_schema(false); + let batch1 = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(StringArray::from(vec!["c1", "c1", "c1"])), + Arc::new(StringArray::from(vec!["s1", "s1", "s1"])), + Arc::new(StringArray::from(vec!["k1", "k2", "k3"])), + Arc::new(Int32Array::from(vec![1, 5, 3])), + Arc::new(Int32Array::from(vec![10, 20, 30])), + ], + ) + .unwrap(); + let batch2 = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(StringArray::from(vec!["c2", "c2", "c2"])), + Arc::new(StringArray::from(vec!["s2", "s2", "s2"])), + Arc::new(StringArray::from(vec!["k1", "k2", "k4"])), + Arc::new(Int32Array::from(vec![4, 2, 6])), + Arc::new(Int32Array::from(vec![40, 50, 60])), + ], + ) + .unwrap(); + + let configs = create_configs("OVERWRITE_WITH_LATEST", true, Some("ts")); + let merger = RecordBatchMerger::new(schema.clone(), Arc::new(configs)); + let batches = RecordBatches::new_with_data_batches([batch1, batch2]); + + let direct_merged = merger.merge_record_batches(batches.clone()).unwrap(); + let trait_merged = ::merge(&merger, batches).unwrap(); + + assert_eq!(trait_merged.schema(), direct_merged.schema()); + assert_eq!( + get_sorted_rows(&trait_merged), + get_sorted_rows(&direct_merged) + ); + } } diff --git a/crates/core/src/table/validation.rs b/crates/core/src/table/validation.rs index 27f0d9d97..37fdd2930 100644 --- a/crates/core/src/table/validation.rs +++ b/crates/core/src/table/validation.rs @@ -25,7 +25,7 @@ use crate::config::table::HudiTableConfig::{ BaseFileFormat, BasePath, DropsPartitionFields, TableVersion, TimelineLayoutVersion, }; use crate::error::CoreError; -use crate::merge::record_merger::RecordMerger; +use crate::merge::validate_configs as validate_record_merger_configs; use crate::util::path::is_metadata_table_path; use std::str::FromStr; use strum::IntoEnumIterator; @@ -82,7 +82,7 @@ pub fn validate_configs(hudi_configs: &HudiConfigs) -> crate::error::Result<()> } } - RecordMerger::validate_configs(hudi_configs)?; + validate_record_merger_configs(hudi_configs)?; Ok(()) }