linliu-code commented on code in PR #669: URL: https://github.com/apache/hudi-rs/pull/669#discussion_r3779925013
########## crates/core/src/schema/batch_evolution.rs: ########## @@ -0,0 +1,1103 @@ +/* + * 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. + */ + +//! Ported from the merge-on-read reader. Nothing consumes it yet, so its +//! items are unreachable from the crate's call graph until the reader wires in. +#![allow(dead_code)] + +//! Batch-level schema-evolution projector. +//! +//! Equivalent of gold's record rewrite (`HoodieAvroUtils.rewriteRecordWithNewSchema`, +//! avro log path) and cast projection (`HoodieParquetFileFormatHelper.generateUnsafeProjection`, +//! parquet base path): reorder columns by name, null-fill added columns, cast +//! promoted types. Gold-parity cast rules: +//! * Float32→Float64: STRING-MEDIATED (both gold paths do this; C6 value-exactness) +//! * numeric→Utf8: Java `String.valueOf` formatting +//! * struct/list/map: recursive +//! * everything else: `arrow_cast::cast` + +use crate::Result; +use crate::error::CoreError; +use arrow_array::{Array, ArrayRef, RecordBatch, StringArray, new_null_array}; +use arrow_schema::{DataType, FieldRef, SchemaRef, TimeUnit}; +use std::sync::Arc; + +/// Microseconds per millisecond — the ÷1000 factor for the NTZ (local-timestamp) +/// micros→millis arithmetic conversion. Mirrors Java `DateTimeUtils.MICROS_PER_MILLIS`. +const MICROS_PER_MILLIS: i64 = 1000; + +/// Project `batch` to `target` schema: reorder by name, null-fill missing +/// nullable columns, evolve types. Identity-cheap when schemas already match. +pub fn project_batch_to_schema(batch: &RecordBatch, target: &SchemaRef) -> Result<RecordBatch> { + if batch.schema() == *target { + return Ok(batch.clone()); + } + let num_rows = batch.num_rows(); + let batch_schema = batch.schema(); + let mut columns: Vec<ArrayRef> = Vec::with_capacity(target.fields().len()); + for tf in target.fields() { + match index_of_ci(&batch_schema, tf.name())? { + Some(idx) => columns.push(evolve_array(batch.column(idx), tf)?), + None => { + if tf.is_nullable() { + columns.push(new_null_array(tf.data_type(), num_rows)); + } else { + return Err(CoreError::Schema(format!( + "evolution: non-nullable column '{}' absent from source batch", + tf.name() + ))); + } + } + } + } + RecordBatch::try_new(target.clone(), columns) + .map_err(|e| CoreError::Schema(format!("evolution: rebuild under target schema: {e}"))) +} + +/// Locate a column by name, preferring an exact match and falling back to a +/// case-insensitive match (gold/Spark resolve field names case-insensitively). +/// +/// Returns `Ok(None)` when no field matches (the caller null-fills) and an +/// error when more than one field matches case-insensitively without an exact +/// match — ambiguous, so fail loudly rather than silently picking one. +pub(crate) fn index_of_ci(schema: &arrow_schema::Schema, name: &str) -> Result<Option<usize>> { + if let Ok(idx) = schema.index_of(name) { + return Ok(Some(idx)); + } + let mut found: Option<usize> = None; + for (idx, field) in schema.fields().iter().enumerate() { + if field.name().eq_ignore_ascii_case(name) { + if found.is_some() { + return Err(CoreError::Schema(format!( + "evolution: column '{name}' matches multiple source columns \ + case-insensitively; ambiguous projection" + ))); + } + found = Some(idx); + } + } + Ok(found) +} + +/// True for any nested/container Arrow type the recursion arms care about. +/// Matching variants (List/Struct/Map) are handled by the recursion arms above +/// the guard; this catches everything else (LargeList, FixedSizeList, and any +/// container present on only one side) so it errors instead of silently routing +/// through `arrow_cast`. +fn is_container(dt: &DataType) -> bool { + matches!( + dt, + DataType::List(_) + | DataType::LargeList(_) + | DataType::FixedSizeList(_, _) + | DataType::Struct(_) + | DataType::Map(_, _) + ) +} + +fn evolve_array(src: &ArrayRef, target_field: &FieldRef) -> Result<ArrayRef> { + let st = src.data_type(); + let tt = target_field.data_type(); + if st == tt { + return Ok(src.clone()); + } + match (st, tt) { + // Gold C6: float→double via string round-trip (both gold paths). + (DataType::Float32, DataType::Float64) => { + let s = float_to_java_string_array(src)?; + arrow_cast::cast(&s, &DataType::Float64) + .map_err(|e| CoreError::Schema(format!("evolution f32->f64: {e}"))) + } + // numeric → string with Java String.valueOf formatting. + // Widening an integer is exact, so a direct cast matches Java. Avro Review Comment: Fixed. Moved the `Int32 → Int64` arm (with its own comment) above the `// numeric → string with Java String.valueOf formatting.` header, so that header once again introduces the float/int→`Utf8` arms it describes. ########## crates/core/src/file_group/log_file/content.rs: ########## @@ -39,13 +46,48 @@ use std::sync::Arc; pub struct Decoder { batch_size: usize, hudi_configs: Arc<HudiConfigs>, + /// Predicate to push into a parquet log block, when the caller has decided + /// it is safe to evaluate before the merge. See + /// [`Decoder::with_row_filter`]. + row_filter: Option<RowFilterBuilder>, + /// Schema an Avro block is resolved up to, as Avro JSON. See + /// [`Decoder::with_required_schema`]. + required_schema_json: Option<String>, Review Comment: Fixed — `reader_schema_json` throughout, matching `AvroBlockDecoder::try_new_with_reader`. Both the `Decoder` field/setter and the `LogFileReader` field/setter are renamed, so the two layers no longer disagree about what to call the resolving schema. -- 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]
