yihua commented on code in PR #660: URL: https://github.com/apache/hudi-rs/pull/660#discussion_r3788515992
########## crates/core/src/file_group/reader_v2/resolver.rs: ########## @@ -0,0 +1,908 @@ +/* + * 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. + */ + +//! Derives a [`ReaderContext`] from a table's resolved configs. + +use crate::Result; +use crate::config::HudiConfigs; +use crate::config::error::ConfigError; +use crate::config::read::HudiReadConfig; +use crate::config::table::{BaseFileFormatValue, HudiTableConfig}; +use crate::error::CoreError; +use crate::file_group::reader_v2::buffer::spillable_map; +use crate::file_group::reader_v2::reader_context::{CONFIG_MERGE_TYPE, MergeMode, ReaderContext}; +use crate::file_group::reader_v2::record_context::RecordContext; +use crate::file_group::reader_v2::schema_handler::FileGroupReaderSchemaHandler; +use crate::timeline::selector::InstantRange; +use std::collections::HashMap; + +/// Resolve the MOR reader context from `hudi_configs`, which the caller has +/// already merged read options into. +#[allow(dead_code)] +pub(crate) fn resolve_reader_context( + hudi_configs: &HudiConfigs, + has_log_files: bool, +) -> Result<ReaderContext> { + let table_path: String = hudi_configs.get(HudiTableConfig::BasePath)?.into(); + + // Resolved by `Table::prepare_reader_options` for table-level reads. A + // `FileGroupReader` built straight from a base URI never loads the timeline, + // so the caller must supply it; defaulting here would silently widen the read. + let latest_commit_time: String = hudi_configs + .try_get(HudiReadConfig::EndTimestamp)? + .ok_or_else(|| ConfigError::NotFound(HudiReadConfig::EndTimestamp.as_ref().to_string()))? + .into(); + + let merge_mode = resolve_merge_mode(hudi_configs)?; Review Comment: Fixed: the resolver now tolerates an unsupported merge mode when there are no log files to merge (commit-time ordering stands in for the mode nothing reads), the merging-read error names `hoodie.read.file.group.reader.version=1`, and end-to-end tests pin the COW, read-optimized, and merging cases. ########## 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: Fixed: `unwrap_ordering_values` decodes ArrayWrapper rows as null ordering cells, so those deletes apply in natural order (warn-logged) instead of failing the read, with a regression test. ########## 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})" Review Comment: Fixed: the null branch is exempt from the mixing check and decodes as null cells that merge as natural-order deletes, matching Hudi, with regression tests for the mixed and all-null shapes. ########## 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: Addressed with the thread above: null and ArrayWrapper branches decode per row as natural-order deletes, and only a genuine two-scalar-type mix still errors. ########## crates/core/src/file_group/file_slice.rs: ########## @@ -29,7 +29,20 @@ use std::path::PathBuf; /// a [FileSlice] is a logical group of [BaseFile] and [LogFile]s. #[derive(Clone, Debug)] pub struct FileSlice { - pub base_file: BaseFile, + /// The base file, when the slice has one. + /// + /// A slice written by inserts that went straight to log files has none — + /// Flink ingestion, a bucket index's first write to a bucket, and any + /// merge-on-read file group before its first compaction all produce these. + pub base_file: Option<BaseFile>, Review Comment: Done: the PR description's behaviour-changes section now calls out the `FileSlice::base_file` / `base_file_relative_path()` API break for crates.io consumers upgrading to 0.5. -- 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]
