This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 85f68e5c [core] Normalize Arrow layouts for upserts and data evolution
updates (#1144)
85f68e5c is described below
commit 85f68e5c7d09893cc77cb357607053c6686c4760
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 17 16:35:06 2026 +0800
[core] Normalize Arrow layouts for upserts and data evolution updates
(#1144)
---
crates/paimon/src/table/data_evolution_writer.rs | 78 ++++++++-
crates/paimon/src/table/table_upsert.rs | 134 ++++++++++++++
crates/paimon/src/table/write_batch_normalize.rs | 211 ++++++++++++++++++++++-
3 files changed, 421 insertions(+), 2 deletions(-)
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index a2069eb8..446f7f94 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<ArrayRef> {
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<Option<&str>>, validity: Option<Vec<bool>>| ->
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::<Vec<bool>>);
+ 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::<StructArray>().unwrap();
+ assert!(result.is_null(0));
+ assert_eq!(
+ result
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .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::<Vec<bool>>)),
+ ])
+ .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<ArrayRef>,
diff --git a/crates/paimon/src/table/table_upsert.rs
b/crates/paimon/src/table/table_upsert.rs
index 769a9734..8869e527 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::<StringArray>()
+ .unwrap();
+ let values = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let keep = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let ids = batch
+ .column(3)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .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 ad5966cc..e7de7ec7 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<ArrayRef> {
+ 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::<Result<Vec<_>>>()?;
+ 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::<u8>().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::<StringArray>().unwrap();
+ assert_eq!(result.value_offsets(), &[0, 3, 3, 9]);
+ assert_eq!(result.values().as_ptr(), unsafe {
+ address.cast::<u8>().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::<LargeStringArray>()
+ .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::<StructArray>().unwrap();
+ assert!(row.is_null(0));
+ assert_eq!(
+ row.column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .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::<ListArray>()
+ .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"] {
