yihua commented on code in PR #660:
URL: https://github.com/apache/hudi-rs/pull/660#discussion_r3788530913
##########
crates/core/src/schema/delete.rs:
##########
@@ -37,9 +37,54 @@ static DELETE_RECORD_AVRO_SCHEMA_IN_JSON:
Lazy<Result<JsonValue>> = Lazy::new(||
.map_err(|e| CoreError::Schema(format!("Failed to parse schema to
JSON: {e}")))
});
+/// Union position of `ArrayWrapper`, which carries a list rather than a
scalar.
+/// Nothing orders records by a list, so it is rejected rather than mapped.
+const ARRAY_WRAPPER_POSITION: u32 = 12;
+
+/// Replace a wrapped ordering value with the primitive inside it.
+///
+/// Hudi writes `orderingVal` as a union of per-type wrapper records —
`LongWrapper`
+/// is a record whose single `value` field is a `long`. The Arrow side wants
the
+/// primitive, and [`avro_schema_for_delete_record`] narrows the schema to
match,
+/// so the value has to be unwrapped to agree with it.
+///
+/// The narrowed schema has two branches, `[null, <primitive>]`, so the
surviving
+/// branch is position 1 regardless of where the wrapper sat in the full union.
+///
+/// Records that are already primitives pass through: this runs over decoded
+/// values, and only the wrapper shape is rewritten.
+pub fn unwrap_ordering_value(delete_record: AvroValue) -> Result<AvroValue> {
+ let AvroValue::Record(mut fields) = delete_record else {
+ return Err(CoreError::Schema(
+ "Expected a record for delete record".to_string(),
+ ));
+ };
+ let Some((_, ordering_val)) = fields.get_mut(2) else {
+ return Err(CoreError::Schema(
+ "Delete record has no orderingVal field".to_string(),
+ ));
+ };
+ if let AvroValue::Union(pos, inner) = ordering_val {
+ if *pos == ARRAY_WRAPPER_POSITION {
Review Comment:
Correction: rather than degrading, the union column now passes through whole
and `delete_batch_to_keys_with_ordering` decodes ArrayWrapper per row into the
composite ordering value the merge compares, so a stale composite delete loses
correctly, with regression tests for the decode and the merge outcome.
##########
crates/core/src/file_group/log_file/content.rs:
##########
@@ -17,35 +17,128 @@
* under the License.
*/
use crate::Result;
-use crate::avro_to_arrow::arrow_array_reader::AvroArrowArrayReader;
use crate::config::HudiConfigs;
use crate::error::CoreError;
-use crate::file_group::log_file::avro::AvroDataBlockContentReader;
+use crate::file_group::log_file::avro::AvroBlockDecoder;
use crate::file_group::log_file::log_block::{
BlockMetadataKey, BlockType, LogBlockContent, LogBlockVersion,
};
use crate::file_group::log_file::log_format::LogFormatVersion;
use crate::file_group::record_batches::RecordBatches;
use crate::hfile::{HFileReader, HFileRecord};
-use crate::schema::delete::{avro_schema_for_delete_record,
avro_schema_for_delete_record_list};
-use apache_avro::{Schema as AvroSchema, from_avro_datum};
+use crate::schema::delete::delete_record_list_schema_json;
+use
crate::schema::extended_promotion::record_needs_rewrite_for_extended_promotion;
+use crate::schema::parquet_list_norm::normalize_parquet_metadata;
+use crate::schema::resolver::avro_json_to_arrow_schema;
+use crate::storage::RowFilterBuilder;
+use arrow_array::{Array, ArrayRef, ListArray, RecordBatch, StructArray,
UnionArray};
+use arrow_schema::{DataType, Field, Schema};
use bytes::Bytes;
-use parquet::arrow::arrow_reader::ParquetRecordBatchReader;
+use parquet::arrow::arrow_reader::{
+ ArrowReaderMetadata, ArrowReaderOptions, ParquetRecordBatchReaderBuilder,
+};
+use parquet::file::metadata::ParquetMetaDataReader;
use std::collections::HashMap;
use std::io::{Read, Seek};
use std::sync::Arc;
+/// Turn the wrapped ordering values into a plain column.
+///
+/// Hudi writes `orderingVal` as a union of per-type wrapper records, so a
decode
+/// against that schema yields a union of one-field structs. The merge wants
the
+/// value itself.
+///
+/// A block writes one ordering type, so exactly one branch is populated; that
+/// branch's `value` child is the column. A block mixing branches is rejected
+/// rather than silently reduced to one of them.
+fn unwrap_ordering_values(ordering: &ArrayRef) -> Result<ArrayRef> {
+ let union = ordering
+ .as_any()
+ .downcast_ref::<UnionArray>()
+ .ok_or_else(|| {
+ CoreError::LogBlockError(format!(
+ "Expected orderingVal to be a union, got {}",
+ ordering.data_type()
+ ))
+ })?;
+
+ let mut active: Option<i8> = None;
+ for i in 0..union.len() {
+ let type_id = union.type_id(i);
+ match active {
+ None => active = Some(type_id),
+ Some(seen) if seen == type_id => {}
+ Some(seen) => {
+ return Err(CoreError::LogBlockError(format!(
+ "Delete block mixes ordering types (union branches {seen}
and {type_id})"
+ )));
+ }
+ }
+ }
+ let Some(active) = active else {
+ return Ok(ordering.clone());
+ };
+
+ let child = union.child(active);
+ // Null is a branch like any other; there is nothing to unwrap out of it.
Review Comment:
Correction to the above: ArrayWrapper rows now decode to composite ordering
values (union passthrough) rather than natural order; null branches still
decode as natural-order deletes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
