diff --git a/crates/paimon/src/table/data_evolution_writer.rs b/crates/paimon/src/table/data_evolution_writer.rs index a2069eb81..446f7f94e 100644 --- a/crates/paimon/src/table/data_evolution_writer.rs +++ b/crates/paimon/src/table/data_evolution_writer.rs @@ -1121,7 +1121,12 @@ fn apply_field_updates( matches: &[(usize, usize, usize)], ) -> Result { if update_columns.iter().any(|column| column == path) { - return interleave_updated_leaf(path, original, matched_batches, matches); + let updated = interleave_updated_leaf(path, original, matched_batches, matches)?; + // Readers allow runtime NULLs and relax nested field nullability. + // Restore the declared write layout, validating only visible children, + // before constructing a partial-column file with the table schema. + let target = crate::arrow::paimon_type_to_arrow(write_field.data_type())?; + return super::write_batch_normalize::normalize_write_array(&updated, &target); } let DataType::Row(row) = write_field.data_type() else { @@ -1973,6 +1978,77 @@ mod tests { .contains("Ambiguous data-evolution write path")); } + #[test] + fn whole_row_updates_restore_declared_child_nullability() { + let write_field = DataField::new( + 0, + "value".into(), + DataType::Row(RowType::new(vec![DataField::new( + 1, + "text".into(), + DataType::VarChar(crate::spec::VarCharType::with_nullable(false, 100).unwrap()), + )])), + ); + let runtime_fields = vec![arrow_schema::Field::new("text", ArrowDataType::Utf8, true)]; + let row = |values: Vec>, validity: Option>| -> ArrayRef { + Arc::new(StructArray::new( + runtime_fields.clone().into(), + vec![Arc::new(StringArray::from(values))], + validity.map(NullBuffer::from), + )) + }; + let original = row(vec![Some("old"), Some("keep")], None::>); + let updates = RecordBatch::try_from_iter([ + ( + "_ROW_ID", + Arc::new(Int64Array::from(vec![0, 1])) as ArrayRef, + ), + ( + "value", + row(vec![None, Some("中文")], Some(vec![false, true])), + ), + ]) + .unwrap(); + let result = apply_field_updates( + &write_field, + "value", + &original, + &["value".into()], + &[updates], + &[(0, 0, 0), (1, 0, 1)], + ) + .unwrap(); + let expected = crate::arrow::paimon_type_to_arrow(write_field.data_type()).unwrap(); + assert_eq!(result.data_type(), &expected); + let result = result.as_any().downcast_ref::().unwrap(); + assert!(result.is_null(0)); + assert_eq!( + result + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(1), + "中文" + ); + // Required children are enforced for a visible ROW, including when + // Arrow's runtime read type had relaxed their nullability. + let invalid = RecordBatch::try_from_iter([ + ("_ROW_ID", Arc::new(Int64Array::from(vec![0])) as ArrayRef), + ("value", row(vec![None], None::>)), + ]) + .unwrap(); + assert!(apply_field_updates( + &write_field, + "value", + &original, + &["value".into()], + &[invalid], + &[(0, 0, 0)] + ) + .is_err()); + } + fn nested_profile_batch( fields: &[DataField], columns: Vec, diff --git a/crates/paimon/src/table/table_upsert.rs b/crates/paimon/src/table/table_upsert.rs index 769a9734f..8869e527e 100644 --- a/crates/paimon/src/table/table_upsert.rs +++ b/crates/paimon/src/table/table_upsert.rs @@ -406,6 +406,140 @@ mod tests { use super::*; use crate::spec::{DataType as PaimonType, DecimalType, TimestampType, VarCharType}; + #[tokio::test] + async fn upsert_normalizes_string_offsets_before_matching_and_writing() { + use crate::catalog::Identifier; + use crate::io::FileIOBuilder; + use crate::spec::{IntType, Schema, TableSchema}; + use crate::table::TableCommit; + use arrow_array::{Array, Int32Array, LargeStringArray, StringArray}; + + let schema = Schema::builder() + .column("key", PaimonType::VarChar(VarCharType::default())) + .column("value", PaimonType::VarChar(VarCharType::default())) + .column("keep", PaimonType::Int(IntType::new())) + .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") + .build() + .unwrap(); + let table = Table::new( + FileIOBuilder::new("memory").build().unwrap(), + Identifier::new("db", "upsert"), + "memory:/upsert".into(), + TableSchema::new(0, &schema), + None, + ); + let mut writer = table + .new_write_builder() + .with_commit_user("test") + .unwrap() + .new_write() + .unwrap(); + let initial = RecordBatch::try_from_iter([ + ( + "key", + Arc::new(StringArray::from(vec![Some("中文"), Some("中文"), None])) as ArrayRef, + ), + ( + "value", + Arc::new(StringArray::from(vec!["old"; 3])) as ArrayRef, + ), + ( + "keep", + Arc::new(Int32Array::from(vec![10, 11, 12])) as ArrayRef, + ), + ]) + .unwrap(); + writer.write_arrow_batch(&initial).await.unwrap(); + TableCommit::new(table.clone(), "seed".into()) + .commit(writer.prepare_commit().await.unwrap()) + .await + .unwrap(); + writer.close().await; + + let source = RecordBatch::try_from_iter([ + ( + "value", + Arc::new(LargeStringArray::from(vec![ + Some("discarded"), + Some(""), + None, + Some("final"), + ])) as ArrayRef, + ), + ( + "key", + Arc::new(LargeStringArray::from(vec![ + Some("中文"), + Some("new"), + None, + Some("中文"), + ])) as ArrayRef, + ), + ]) + .unwrap(); + let mut updater = table.new_write_builder().new_update().unwrap(); + updater.with_update_type(vec!["value".into()]).unwrap(); + let messages = updater + .upsert_by_arrow_with_key( + vec![source.slice(0, 1), source.slice(1, 3)], + vec!["key".into()], + ) + .await + .unwrap(); + TableCommit::new(table.clone(), "upsert".into()) + .commit(messages) + .await + .unwrap(); + let mut reader = table.new_read_builder(); + reader + .with_projection(&["key", "value", "keep", "_ROW_ID"]) + .unwrap(); + let plan = reader.new_scan().plan().await.unwrap(); + let mut batches = reader.new_read().unwrap().to_arrow(plan.splits()).unwrap(); + let mut rows = Vec::new(); + while let Some(batch) = batches.try_next().await.unwrap() { + let keys = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let values = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + let keep = batch + .column(2) + .as_any() + .downcast_ref::() + .unwrap(); + let ids = batch + .column(3) + .as_any() + .downcast_ref::() + .unwrap(); + rows.extend((0..batch.num_rows()).map(|row| { + ( + ids.value(row), + (!keys.is_null(row)).then(|| keys.value(row).to_string()), + (!values.is_null(row)).then(|| values.value(row).to_string()), + (!keep.is_null(row)).then(|| keep.value(row)), + ) + })); + } + rows.sort(); + assert_eq!( + rows, + vec![ + (0, Some("中文".into()), Some("final".into()), Some(10)), + (1, Some("中文".into()), Some("final".into()), Some(11)), + (2, None, None, Some(12)), + (3, Some("new".into()), Some("".into()), None), + ] + ); + } + #[test] fn partition_identity_ignores_java_null_variable_storage() { let fields = vec![ diff --git a/crates/paimon/src/table/write_batch_normalize.rs b/crates/paimon/src/table/write_batch_normalize.rs index ad5966cc4..e7de7ec7e 100644 --- a/crates/paimon/src/table/write_batch_normalize.rs +++ b/crates/paimon/src/table/write_batch_normalize.rs @@ -26,7 +26,7 @@ use crate::{Error, Result}; use arrow_array::{ Array, ArrayRef, BinaryArray, Decimal128Array, FixedSizeBinaryArray, FixedSizeListArray, - ListArray, MapArray, StringArray, StructArray, + LargeStringArray, ListArray, MapArray, StringArray, StructArray, }; use arrow_buffer::{NullBuffer, OffsetBuffer, ScalarBuffer}; use arrow_schema::{DataType, Field}; @@ -230,6 +230,11 @@ fn normalize_visible_array( } match (array.data_type(), expected) { + // Java has one STRING value representation. Arrow offset widths do + // not change the logical values, including in keys and nested fields. + (DataType::LargeUtf8, DataType::Utf8) => { + normalize_large_string(array.as_any().downcast_ref().unwrap()) + } (DataType::FixedSizeBinary(_), DataType::Binary) => { let fixed = array .as_any() @@ -343,6 +348,34 @@ fn normalize_visible_array( } } +/// Narrow only the visible byte range. A small slice may start beyond i32::MAX +/// in its backing buffer, so casting the original absolute offsets is wrong. +/// The value buffer remains shared and only the offset buffer is rebuilt. +fn normalize_large_string(array: &LargeStringArray) -> Result { + let source = array.value_offsets(); + let start = source[0]; + let end = source[source.len() - 1]; + let offsets = source + .iter() + .map(|offset| { + i32::try_from(offset - start).map_err(|_| Error::DataInvalid { + message: "Arrow STRING byte range exceeds 32-bit offsets".into(), + source: None, + }) + }) + .collect::>>()?; + let values = array + .values() + .slice_with_length(start as usize, (end - start) as usize); + StringArray::try_new( + OffsetBuffer::new(offsets.into()), + values, + array.nulls().cloned(), + ) + .map(|array| Arc::new(array) as ArrayRef) + .map_err(normalization_error) +} + /// Keep only children referenced by visible parents. Ordinary inputs and /// slices share their payload buffers; only disjoint ranges across NULL /// parents need concatenation. Hidden children must not fail NOT NULL checks @@ -421,6 +454,182 @@ mod tests { use arrow_buffer::{OffsetBuffer, ScalarBuffer}; use arrow_schema::Field; + #[cfg(all(unix, target_pointer_width = "64"))] + #[test] + fn large_string_slices_after_two_gib_keep_only_the_visible_byte_range() { + use arrow_array::LargeStringArray; + use arrow_buffer::Buffer; + use std::ptr::NonNull; + + struct Mapping { + address: usize, + len: usize, + } + impl Drop for Mapping { + fn drop(&mut self) { + // SAFETY: the mapping owns this mmap region until the last + // Arrow buffer using it is dropped. + unsafe { libc::munmap(self.address as *mut libc::c_void, self.len) }; + } + } + let start = i32::MAX as usize + 17; + let tail = "abc中文tail".as_bytes(); + let len = start + tail.len(); + // Reserve virtual memory without allocating/touching a 2 GiB payload. + // Only the small visible tail is initialized and read by this test. + let address = unsafe { + libc::mmap( + std::ptr::null_mut(), + len, + libc::PROT_READ | libc::PROT_WRITE, + libc::MAP_PRIVATE | libc::MAP_ANON, + -1, + 0, + ) + }; + assert_ne!(address, libc::MAP_FAILED); + let owner = Arc::new(Mapping { + address: address as usize, + len, + }); + // SAFETY: mmap returned len initialized bytes; the tail fits within + // the mapping and all writes finish before publishing the buffer. + let values = unsafe { + std::ptr::copy_nonoverlapping( + tail.as_ptr(), + address.cast::().add(start), + tail.len(), + ); + Buffer::from_custom_allocation(NonNull::new(address.cast()).unwrap(), len, owner) + }; + let start = start as i64; + let input = LargeStringArray::new( + OffsetBuffer::new(vec![start, start + 3, start + 3, start + 9, start + 13].into()), + values, + Some(NullBuffer::from(vec![true, false, true, true])), + ); + let input: ArrayRef = Arc::new(input.slice(0, 3)); + let result = normalize_write_array(&input, &DataType::Utf8).unwrap(); + assert_eq!( + result.to_data(), + StringArray::from(vec![Some("abc"), None, Some("中文")]).to_data() + ); + let result = result.as_any().downcast_ref::().unwrap(); + assert_eq!(result.value_offsets(), &[0, 3, 3, 9]); + assert_eq!(result.values().as_ptr(), unsafe { + address.cast::().add(start as usize) + }); + let empty = normalize_write_array(&input.slice(3, 0), &DataType::Utf8).unwrap(); + assert!(empty.is_empty()); + // A visible range larger than i32::MAX still cannot be represented. + // SAFETY: the mapped bytes are zeroes followed by a valid UTF-8 tail; + // offsets and NULL bitmap are valid. Skip scanning the 2 GiB prefix. + let too_wide: ArrayRef = Arc::new(unsafe { + LargeStringArray::new_unchecked( + OffsetBuffer::new(vec![0, start + 13].into()), + input + .as_any() + .downcast_ref::() + .unwrap() + .values() + .clone(), + Some(NullBuffer::from(vec![false])), + ) + }); + assert!(normalize_write_array(&too_wide, &DataType::Utf8) + .unwrap_err() + .to_string() + .contains("32-bit offsets")); + } + + #[test] + fn binary_offset_widths_are_not_implicitly_converted() { + let small: ArrayRef = Arc::new(BinaryArray::from(vec![Some(b"\xff\0".as_slice()), None])); + let large: ArrayRef = Arc::new(LargeBinaryArray::from(vec![ + Some(b"\xff\0".as_slice()), + None, + ])); + assert!(normalize_write_array(&small, &DataType::LargeBinary).is_err()); + assert!(normalize_write_array(&large, &DataType::Binary).is_err()); + } + + #[test] + fn large_strings_preserve_slices_unicode_and_nulls() { + let input: ArrayRef = Arc::new( + arrow_array::LargeStringArray::from(vec![ + Some("outside"), + Some("中文"), + None, + Some(""), + Some("last"), + ]) + .slice(1, 4), + ); + let result = normalize_write_array(&input, &DataType::Utf8).unwrap(); + assert_eq!( + result.to_data(), + StringArray::from(vec![Some("中文"), None, Some(""), Some("last")]).to_data() + ); + let empty = normalize_write_array(&input.slice(0, 0), &DataType::Utf8).unwrap(); + assert!(empty.is_empty()); + assert_eq!(empty.data_type(), &DataType::Utf8); + // Layout compatibility must not become an implicit value cast. + assert!(normalize_write_array(&input, &DataType::Int32).is_err()); + assert!(normalize_write_array(&input, &DataType::Binary).is_err()); + } + + #[test] + fn large_strings_inside_rows_and_arrays_keep_visible_nullability_checks() { + let text: ArrayRef = Arc::new(arrow_array::LargeStringArray::from(vec![ + None, + Some("中文"), + None, + ])); + let input = Arc::new(StructArray::new( + vec![Field::new("text", DataType::LargeUtf8, true)].into(), + vec![text], + Some(NullBuffer::from(vec![false, true, true])), + )) as ArrayRef; + let target = DataType::Struct(vec![Field::new("text", DataType::Utf8, false)].into()); + assert!(normalize_write_array(&input, &target).is_err()); + // The first child's NULL is hidden by a NULL ROW, so it is allowed. + let normalized = normalize_write_array(&input.slice(0, 2), &target).unwrap(); + let row = normalized.as_any().downcast_ref::().unwrap(); + assert!(row.is_null(0)); + assert_eq!( + row.column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(1), + "中文" + ); + let list = Arc::new( + ListArray::new( + Arc::new(Field::new("item", DataType::LargeUtf8, true)), + OffsetBuffer::new(vec![0, 1, 3].into()), + Arc::new(arrow_array::LargeStringArray::from(vec![ + Some("outside"), + Some("中文"), + None, + ])), + None, + ) + .slice(1, 1), + ) as ArrayRef; + let expected = DataType::List(Arc::new(Field::new("element", DataType::Utf8, true))); + let list = normalize_write_array(&list, &expected).unwrap(); + assert_eq!(list.data_type(), &expected); + assert_eq!( + list.as_any() + .downcast_ref::() + .unwrap() + .value(0) + .to_data(), + StringArray::from(vec![Some("中文"), None]).to_data() + ); + } + #[test] fn vector_child_aliases_preserve_sliced_values_and_nulls() { for name in ["item", "", "element"] {