hudi-agent commented on code in PR #669:
URL: https://github.com/apache/hudi-rs/pull/669#discussion_r3771771928
##########
crates/core/src/file_group/log_file/content.rs:
##########
@@ -110,23 +152,48 @@ impl Decoder {
) -> Result<RecordBatches> {
Decoder::validate_log_block_version(&mut reader)?;
- let writer_schema =
header.get(&BlockMetadataKey::Schema).ok_or_else(|| {
+ let writer_schema_json =
header.get(&BlockMetadataKey::Schema).ok_or_else(|| {
CoreError::LogBlockError("Schema not found in block
header".to_string())
})?;
- let writer_schema = Arc::new(AvroSchema::parse_str(writer_schema)?);
let mut record_count_buf = [0u8; 4];
reader.read_exact(&mut record_count_buf)?;
let record_count = u32::from_be_bytes(record_count_buf);
- let record_content_reader =
- AvroDataBlockContentReader::new(reader, writer_schema.as_ref(),
record_count);
- let mut avro_arrow_array_reader =
- AvroArrowArrayReader::try_new(record_content_reader,
writer_schema.as_ref())?;
+ // A partial-update block carries only the columns that were written,
and
+ // the merge needs to know which those are. Resolving it up to the
table
+ // schema would fabricate the rest, so it decodes against its own
schema.
+ let is_partial = header.contains_key(&BlockMetadataKey::IsPartial);
+ let reader_schema_json = if is_partial {
+ None
+ } else {
+ self.required_schema_json.as_deref()
+ };
+ let mut decoder = AvroBlockDecoder::try_new_with_reader(
Review Comment:
🤖 When `required_schema_json` promotes a float→double column, resolving the
block through the Avro reader schema uses arrow-avro's native widening (1.1f32
→ 1.100000023841858), whereas `batch_evolution` deliberately string-mediates
Float32→Float64 for base files to match the Java gold. Would base-file rows and
log-updated rows for that column then carry different doubles (and diverge from
Java Hudi)? The test's int→long promotion is exact, so it wouldn't surface this.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
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:
🤖 nit: the pre-existing `// numeric → string with Java String.valueOf
formatting.` comment immediately precedes this arm, so it now reads as the
annotation for the int→long widening rather than for the string-conversion arms
below. Could you move the new arm (and its own comment) to before that section
header so the comment still points at the float/int→Utf8 block?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
crates/core/src/file_group/log_file/content.rs:
##########
@@ -110,23 +152,48 @@ impl Decoder {
) -> Result<RecordBatches> {
Decoder::validate_log_block_version(&mut reader)?;
- let writer_schema =
header.get(&BlockMetadataKey::Schema).ok_or_else(|| {
+ let writer_schema_json =
header.get(&BlockMetadataKey::Schema).ok_or_else(|| {
CoreError::LogBlockError("Schema not found in block
header".to_string())
})?;
- let writer_schema = Arc::new(AvroSchema::parse_str(writer_schema)?);
let mut record_count_buf = [0u8; 4];
reader.read_exact(&mut record_count_buf)?;
let record_count = u32::from_be_bytes(record_count_buf);
- let record_content_reader =
- AvroDataBlockContentReader::new(reader, writer_schema.as_ref(),
record_count);
- let mut avro_arrow_array_reader =
- AvroArrowArrayReader::try_new(record_content_reader,
writer_schema.as_ref())?;
+ // A partial-update block carries only the columns that were written,
and
+ // the merge needs to know which those are. Resolving it up to the
table
+ // schema would fabricate the rest, so it decodes against its own
schema.
+ let is_partial = header.contains_key(&BlockMetadataKey::IsPartial);
Review Comment:
🤖 Worth digging into, because the merge stage doesn't appear to backstop
this. `schema_is_partial` (key_based.rs:1211) detects partial-ness by missing
target fields, so a *full* block that got mis-decoded writer-only here would
still carry every target field and be treated as full at merge — meaning the
int→long promotion never happens and the wrong type reaches merge silently
rather than being caught. So the presence-vs-value distinction matters. Do we
know whether Java's writer ever actually emits an explicit `IS_PARTIAL=false`
header, or does it only add the header for true partials? That determines
whether this is reachable in practice.
##########
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:
🤖 nit: the same concept is `reader_schema_json` in
`AvroBlockDecoder::try_new_with_reader` (Avro's own name for the resolving
schema) but `required_schema_json` here and in `LogFileReader`. Could you pick
one name — `reader_schema_json` matches the Avro spec terminology and the
lower-level API — and use it consistently across both layers?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
