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 952f6b81 [core] Complete nested updates and partition-aware temporal
upserts (#973)
952f6b81 is described below
commit 952f6b81205be273be33fcd481802ed111b048ca
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Sep 27 22:57:02 2026 +0800
[core] Complete nested updates and partition-aware temporal upserts (#973)
---
crates/paimon/src/table/table_upsert.rs | 122 +++++-
crates/paimon/src/table/update_input.rs | 510 ++++++++++++++++++++++-
crates/paimon/src/table/upsert_key_matcher.rs | 61 +++
crates/paimon/src/table/write_batch_normalize.rs | 34 +-
crates/paimon/tests/table_update_nested_test.rs | 439 +++++++++++++++++++
5 files changed, 1142 insertions(+), 24 deletions(-)
diff --git a/crates/paimon/src/table/table_upsert.rs
b/crates/paimon/src/table/table_upsert.rs
index 3a62311a..aa8ee45f 100644
--- a/crates/paimon/src/table/table_upsert.rs
+++ b/crates/paimon/src/table/table_upsert.rs
@@ -17,6 +17,8 @@
//! Table-level append-only upsert by user-specified keys.
+use std::borrow::Cow;
+use std::collections::HashSet;
use std::sync::Arc;
use arrow_array::{ArrayRef, Int64Array, RecordBatch, UInt32Array};
@@ -25,7 +27,11 @@ use arrow_select::{concat::concat_batches, take::take};
use futures::TryStreamExt;
use super::upsert_key_matcher::UpsertKeyMatcher;
-use crate::spec::CoreOptions;
+use super::write_batch_normalize::normalize_write_array;
+use crate::spec::{
+ batch_to_serialized_bytes, extract_datum, BinaryRow, BinaryRowBuilder,
CoreOptions, DataField,
+};
+use crate::table::source::{DataSplit, Plan};
use crate::table::{CommitMessage, Table};
const ROW_ID: &str = "_ROW_ID";
@@ -55,6 +61,19 @@ fn selected_rows(batch: &RecordBatch, rows: &[usize]) ->
crate::Result<RecordBat
.map_err(|error| invalid(format!("cannot build selected upsert rows:
{error}")))
}
+// Java may reserve variable-length storage for NULL timestamps/decimals.
+// Compare logical partitions using one encoding, independent of that padding.
+fn partition_key(row: &BinaryRow, fields: &[DataField]) ->
crate::Result<Vec<u8>> {
+ let mut builder = BinaryRowBuilder::new(fields.len() as i32);
+ for (index, field) in fields.iter().enumerate() {
+ match extract_datum(row, index, field.data_type())? {
+ Some(value) => builder.write_datum(index, &value,
field.data_type()),
+ None => builder.set_null_at(index),
+ }
+ }
+ Ok(builder.build().to_serialized_bytes())
+}
+
/// Upsert full Arrow rows into a data-evolution table without primary keys.
/// Internal executor for `TableUpdate::upsert_by_arrow_with_key`. Existing
/// keys are updated by row ID; new keys are appended.
@@ -119,7 +138,8 @@ impl TableUpsert {
}
/// Add full rows. Column order may differ from the table schema; names and
- /// Arrow types must agree. Multiple batches form one logical upsert input.
+ /// Arrow layouts follow the same normalization as ordinary writes.
Multiple
+ /// batches form one logical upsert input.
pub(super) fn add_batch(&mut self, batch: RecordBatch) ->
crate::Result<()> {
let target =
crate::arrow::build_target_arrow_schema(self.table.schema().fields())?;
if batch.num_columns() != target.fields().len() {
@@ -130,15 +150,11 @@ impl TableUpsert {
let column = batch
.column_by_name(field.name())
.ok_or_else(|| invalid(format!("missing upsert column '{}'",
field.name())))?;
- if column.data_type() != field.data_type() {
- return Err(invalid(format!(
- "upsert column '{}' type differs from table: {:?} != {:?}",
- field.name(),
- column.data_type(),
- field.data_type()
- )));
- }
- columns.push(column.clone());
+ columns.push(
+ normalize_write_array(column,
field.data_type()).map_err(|error| {
+ invalid(format!("Invalid upsert column '{}': {error}",
field.name()))
+ })?,
+ );
}
let ordered = RecordBatch::try_new(target, columns)
.map_err(|error| invalid(format!("cannot order upsert columns:
{error}")))?;
@@ -146,6 +162,40 @@ impl TableUpsert {
Ok(())
}
+ /// Match only the source partitions, as PyPaimon does. Filter planned
splits
+ /// after global row IDs have been assigned so pruning cannot renumber
rows.
+ fn matching_splits<'a>(
+ &self,
+ source: &RecordBatch,
+ plan: &'a Plan,
+ ) -> crate::Result<Cow<'a, [DataSplit]>> {
+ let schema = self.table.schema();
+ if schema.partition_keys().is_empty() {
+ return Ok(Cow::Borrowed(plan.splits()));
+ }
+ let indices = schema
+ .partition_keys()
+ .iter()
+ .map(|name| {
+ source
+ .schema()
+ .index_of(name)
+ .map_err(|error| invalid(error.to_string()))
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ let partitions: HashSet<_> = batch_to_serialized_bytes(source,
&indices, schema.fields())?
+ .into_iter()
+ .collect();
+ let fields = schema.partition_fields();
+ let mut splits = Vec::new();
+ for split in plan.splits() {
+ if partitions.contains(&partition_key(split.partition(),
&fields)?) {
+ splits.push(split.clone());
+ }
+ }
+ Ok(Cow::Owned(splits))
+ }
+
/// Match source keys against the target snapshot, stage row-ID updates and
/// new rows, then return both sets of commit messages as one operation.
#[must_use = "commit messages must be passed to TableCommit"]
@@ -167,7 +217,8 @@ impl TableUpsert {
read_builder.with_projection(&projection)?;
let plan = read_builder.new_scan().plan().await?;
let read = read_builder.new_read()?;
- let mut stream = read.to_arrow(plan.splits())?;
+ let splits = self.matching_splits(&source, &plan)?;
+ let mut stream = read.to_arrow(splits.as_ref())?;
while let Some(batch) = stream.try_next().await? {
matcher.add_existing_batch(&batch)?;
}
@@ -221,3 +272,50 @@ impl TableUpsert {
result.map(|()| messages)
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::{DataType as PaimonType, DecimalType, TimestampType,
VarCharType};
+
+ #[test]
+ fn partition_identity_ignores_java_null_variable_storage() {
+ let fields = vec![
+ DataField::new(
+ 0,
+ "ts".into(),
+ PaimonType::Timestamp(TimestampType::new(6).unwrap()),
+ ),
+ DataField::new(
+ 1,
+ "amount".into(),
+ PaimonType::Decimal(DecimalType::new(20, 2).unwrap()),
+ ),
+ DataField::new(
+ 2,
+ "label".into(),
+ PaimonType::VarChar(VarCharType::new(100).unwrap()),
+ ),
+ ];
+ let mut builder = BinaryRowBuilder::new(3);
+ builder.set_null_at(0);
+ builder.set_null_at(1);
+ builder.write_string(2, "after-null-fields");
+ let canonical = builder.build();
+ let mut java = canonical.data()[..32].to_vec();
+ // BinaryWriter.writeTimestamp(NULL, 6) reserves 8 bytes;
+ // writeDecimal(NULL, 20) reserves another 16, even though both are
NULL.
+ java.extend_from_slice(&[0; 24]);
+ java.extend_from_slice(&canonical.data()[32..]);
+ java[8..16].copy_from_slice(&(32_u64 << 32).to_le_bytes());
+ java[16..24].copy_from_slice(&(40_u64 << 32).to_le_bytes());
+ let offset_and_size =
u64::from_le_bytes(canonical.data()[24..32].try_into().unwrap());
+ java[24..32].copy_from_slice(&(offset_and_size + (24_u64 <<
32)).to_le_bytes());
+ let java = BinaryRow::from_bytes(3, java);
+ assert_ne!(java.to_serialized_bytes(),
canonical.to_serialized_bytes());
+ assert_eq!(
+ partition_key(&java, &fields).unwrap(),
+ canonical.to_serialized_bytes()
+ );
+ }
+}
diff --git a/crates/paimon/src/table/update_input.rs
b/crates/paimon/src/table/update_input.rs
index 803a4660..d68de118 100644
--- a/crates/paimon/src/table/update_input.rs
+++ b/crates/paimon/src/table/update_input.rs
@@ -19,6 +19,7 @@
use std::sync::Arc;
+use arrow_array::cast::AsArray;
use arrow_array::{
new_null_array, Array, ArrayRef, FixedSizeListArray, Float32Array,
Float64Array,
GenericListArray, Int64Array, LargeListArray, ListArray, MapArray,
OffsetSizeTrait,
@@ -130,8 +131,24 @@ pub(super) fn cast_update_value(
mode: CastMode,
) -> crate::Result<ArrayRef> {
if mode == CastMode::RowUpdate {
- cast_values(array, target, CastMode::Assignment)
- .or_else(|_| cast_values(array, target, CastMode::Constructor))
+ cast_values(array, target, CastMode::Assignment).or_else(|_| {
+ validate_map_constructor_input(array, target)?;
+ let decoded = decode_dictionary(array)?;
+ // Python converts lists of pairs to tuples only for a top-level
MAP.
+ if let DataType::Map(field, sorted) = target {
+ if matches!(
+ decoded.data_type(),
+ DataType::List(_)
+ | DataType::LargeList(_)
+ | DataType::ListView(_)
+ | DataType::LargeListView(_)
+ | DataType::FixedSizeList(..)
+ ) {
+ return constructor_map(&decoded, field, *sorted);
+ }
+ }
+ cast_values(array, target, CastMode::Constructor)
+ })
} else {
cast_values(array, target, mode)
}
@@ -169,9 +186,26 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode:
CastMode) -> crate::Re
"Run-end encoded update inputs are not supported by PyArrow casts",
));
}
+ if mode == CastMode::Constructor {
+ match (array.data_type(), target) {
+ (DataType::Struct(_), DataType::Map(field, sorted)) => {
+ return constructor_map(array, field, *sorted);
+ }
+ (DataType::Map(..), DataType::Struct(fields)) => {
+ return constructor_struct_from_map(
+ array.as_any().downcast_ref::<MapArray>().unwrap(),
+ fields,
+ );
+ }
+ _ => {}
+ }
+ }
let nested: Option<ArrayRef> = match (array.data_type(), target) {
(DataType::Struct(_), DataType::Struct(fields)) => {
let input = array.as_any().downcast_ref::<StructArray>().unwrap();
+ if mode == CastMode::Assignment {
+ validate_struct_cast_fields(input.fields(), fields)?;
+ }
let columns = fields
.iter()
.map(|field| match input.column_by_name(field.name()) {
@@ -289,6 +323,235 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode:
CastMode) -> crate::Re
cast_primitive(array, target, mode)
}
+// PyPaimon's row-ID fallback reconstructs Python dicts only after a
whole-column
+// safe cast fails. At the top level it rejects ambiguous schema-less null
keys.
+fn validate_map_constructor_input(array: &ArrayRef, target: &DataType) ->
crate::Result<()> {
+ if matches!(target, DataType::Map(..)) {
+ if let Some(input) = array.as_any().downcast_ref::<StructArray>() {
+ if input.columns().iter().any(|column| {
+ column.logical_nulls().is_some_and(|nulls| {
+ (0..input.len()).any(|row| input.is_valid(row) &&
nulls.is_null(row))
+ })
+ }) {
+ return Err(invalid(
+ "Cannot coerce schema-less dict input with null values to
map type; pass an explicit map-typed array or list-of-pairs instead",
+ ));
+ }
+ }
+ }
+ Ok(())
+}
+
+fn validate_struct_cast_fields(
+ source: &arrow_schema::Fields,
+ target: &arrow_schema::Fields,
+) -> crate::Result<()> {
+ let mut previous = None;
+ for field in target {
+ if let Some(index) = source.iter().position(|f| f.name() ==
field.name()) {
+ if previous.is_some_and(|previous| index <= previous) {
+ return Err(invalid(
+ "Struct fields are in the wrong order for a safe cast",
+ ));
+ }
+ if source[index].is_nullable() && !field.is_nullable() {
+ return Err(invalid(
+ "Cannot safely cast a nullable Struct field to
non-nullable",
+ ));
+ }
+ previous = Some(index);
+ }
+ }
+ Ok(())
+}
+
+fn constructor_struct_from_map(
+ input: &MapArray,
+ fields: &arrow_schema::Fields,
+) -> crate::Result<ArrayRef> {
+ let keys = decode_dictionary(input.keys())?;
+ let columns = fields
+ .iter()
+ .enumerate()
+ .map(|(field_index, field)| {
+ let indices = (0..input.len())
+ .map(|row| {
+ let range = input.value_offsets();
+ if input.is_null(row) || field_index >= (range[row + 1] -
range[row]) as usize {
+ return Ok(None);
+ }
+ let index = range[row] as usize + field_index;
+ let key = match keys.data_type() {
+ DataType::Utf8 =>
keys.as_string::<i32>().value(index).as_bytes(),
+ DataType::LargeUtf8 =>
keys.as_string::<i64>().value(index).as_bytes(),
+ DataType::Utf8View =>
keys.as_string_view().value(index).as_bytes(),
+ DataType::Binary =>
keys.as_binary::<i32>().value(index),
+ DataType::LargeBinary =>
keys.as_binary::<i64>().value(index),
+ DataType::BinaryView =>
keys.as_binary_view().value(index),
+ DataType::FixedSizeBinary(_) =>
keys.as_fixed_size_binary().value(index),
+ _ => {
+ return Err(invalid(
+ "Expected a string or bytes field name in
MAP-to-ROW input",
+ ))
+ }
+ };
+ if key != field.name().as_bytes() {
+ return Err(invalid(format!(
+ "Expected field name '{}' in MAP-to-ROW input",
+ field.name()
+ )));
+ }
+ Ok(Some(index as u64))
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ // Missing fields, NULL parents and trailing entries do not
participate
+ // in conversion, matching PyArrow's ordered pair constructor for
ROW.
+ let values = arrow_select::take::take(
+ input.values().as_ref(),
+ &arrow_array::UInt64Array::from(indices),
+ None,
+ )
+ .map_err(|error| invalid(error.to_string()))?;
+ cast_values(&values, field.data_type(), CastMode::Constructor)
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ Ok(Arc::new(
+ StructArray::try_new_with_length(
+ fields.clone(),
+ columns,
+ input.nulls().cloned(),
+ input.len(),
+ )
+ .map_err(|error| invalid(error.to_string()))?,
+ ))
+}
+
+fn constructor_map(
+ input: &ArrayRef,
+ entries_field: &arrow_schema::FieldRef,
+ sorted: bool,
+) -> crate::Result<ArrayRef> {
+ let DataType::Struct(fields) = entries_field.data_type() else {
+ return Err(invalid("Map entries must have a struct type"));
+ };
+ let mut offsets = vec![0_i32];
+ let (keys, values) = if let Some(input) =
input.as_any().downcast_ref::<StructArray>() {
+ let mut indices = Vec::new();
+ let mut keys = Vec::new();
+ for row in 0..input.len() {
+ if input.is_valid(row) {
+ for (index, field) in input.fields().iter().enumerate() {
+ keys.push(field.name().as_str());
+ indices.push((index, row));
+ }
+ }
+ offsets.push(i32::try_from(indices.len()).map_err(|_| invalid("Map
offset overflow"))?);
+ }
+ let columns = input
+ .columns()
+ .iter()
+ .map(|column| {
+ let visible = constructor_child(column, input.nulls(), 1,
CastMode::Constructor)?;
+ cast_values(&visible, fields[1].data_type(),
CastMode::Constructor)
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ let values = if columns.is_empty() {
+ arrow_array::new_empty_array(fields[1].data_type())
+ } else {
+ let columns = columns
+ .iter()
+ .map(|column| column.as_ref())
+ .collect::<Vec<_>>();
+ arrow_select::interleave::interleave(&columns, &indices)
+ .map_err(|error| invalid(error.to_string()))?
+ };
+ (Arc::new(StringArray::from(keys)) as ArrayRef, values)
+ } else {
+ let mut keys = Vec::new();
+ let mut values = Vec::new();
+ for row in 0..input.len() {
+ if input.is_valid(row) {
+ let pairs = constructor_list_value(input, row)?;
+ for index in 0..pairs.len() {
+ if pairs.is_null(index) {
+ return Err(invalid("Map key/value pair must not be
null"));
+ }
+ let pair = constructor_list_value(&pairs, index)?;
+ if pair.len() != 2 {
+ return Err(invalid(
+ "Map key/value pair must contain exactly two
values",
+ ));
+ }
+ keys.push(pair.slice(0, 1));
+ values.push(pair.slice(1, 1));
+ }
+ }
+ offsets.push(i32::try_from(keys.len()).map_err(|_| invalid("Map
offset overflow"))?);
+ }
+ (
+ concat_constructor_values(&keys, fields[0].data_type())?,
+ concat_constructor_values(&values, fields[1].data_type())?,
+ )
+ };
+ let keys = cast_values(&keys, fields[0].data_type(),
CastMode::Constructor)?;
+ let entries = StructArray::try_new(fields.clone(), vec![keys, values],
None)
+ .map_err(|error| invalid(error.to_string()))?;
+ Ok(Arc::new(
+ MapArray::try_new(
+ entries_field.clone(),
+ OffsetBuffer::new(offsets.into()),
+ entries,
+ input.nulls().cloned(),
+ sorted,
+ )
+ .map_err(|error| invalid(error.to_string()))?,
+ ))
+}
+
+fn constructor_list_value(input: &ArrayRef, index: usize) ->
crate::Result<ArrayRef> {
+ match input.data_type() {
+ DataType::List(_) => Ok(input
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .unwrap()
+ .value(index)),
+ DataType::LargeList(_) => Ok(input
+ .as_any()
+ .downcast_ref::<LargeListArray>()
+ .unwrap()
+ .value(index)),
+ DataType::FixedSizeList(..) => Ok(input
+ .as_any()
+ .downcast_ref::<FixedSizeListArray>()
+ .unwrap()
+ .value(index)),
+ DataType::ListView(_) => Ok(input
+ .as_any()
+ .downcast_ref::<arrow_array::ListViewArray>()
+ .unwrap()
+ .value(index)),
+ DataType::LargeListView(_) => Ok(input
+ .as_any()
+ .downcast_ref::<arrow_array::LargeListViewArray>()
+ .unwrap()
+ .value(index)),
+ _ => Err(invalid("Map input must contain key/value pairs")),
+ }
+}
+
+fn concat_constructor_values(values: &[ArrayRef], target: &DataType) ->
crate::Result<ArrayRef> {
+ if values.is_empty() {
+ return Ok(arrow_array::new_empty_array(target));
+ }
+ let values = values
+ .iter()
+ .map(|array| array.as_ref())
+ .collect::<Vec<_>>();
+ let combined =
+ arrow_select::concat::concat(&values).map_err(|error|
invalid(error.to_string()))?;
+ cast_values(&combined, target, CastMode::Constructor)
+}
+
fn cast_list<O: OffsetSizeTrait>(
input: &GenericListArray<O>,
field: &arrow_schema::FieldRef,
@@ -2086,4 +2349,247 @@ mod tests {
assert!(float_to_decimal(value, false, 5, 2).is_err());
}
}
+
+ fn test_map_type(key: DataType, value: DataType) -> DataType {
+ DataType::Map(
+ Arc::new(arrow_schema::Field::new(
+ "entries",
+ DataType::Struct(
+ vec![
+ arrow_schema::Field::new("key", key, false),
+ arrow_schema::Field::new("value", value, true),
+ ]
+ .into(),
+ ),
+ false,
+ )),
+ false,
+ )
+ }
+
+ #[test]
+ fn row_update_dict_to_map_distinguishes_root_and_nested_nulls() {
+ use arrow_schema::Field;
+ let input: ArrayRef = Arc::new(StructArray::new(
+ vec![Field::new("a", DataType::Float64, true)].into(),
+ vec![Arc::new(Float64Array::from(vec![
+ Some(1.5),
+ None,
+ Some(f64::NAN),
+ ]))],
+ Some(arrow_buffer::NullBuffer::from(vec![true, true, false])),
+ ));
+ let target = test_map_type(DataType::Utf8, DataType::Int32);
+ assert!(cast_update_value(&input, &target,
CastMode::RowUpdate).is_err());
+ assert!(cast_assignment(&input.slice(0, 1), &target).is_err());
+ let result = cast_update_value(&input.slice(0, 1), &target,
CastMode::RowUpdate).unwrap();
+ let map = result.as_any().downcast_ref::<MapArray>().unwrap();
+ assert_eq!(map.keys().to_data(),
StringArray::from(vec!["a"]).to_data());
+ assert_eq!(map.values().to_data(),
Int32Array::from(vec![1]).to_data());
+
+ let nested: ArrayRef = Arc::new(StructArray::from(vec![(
+ Arc::new(Field::new("m", input.data_type().clone(), true)),
+ input,
+ )]));
+ let nested_target = DataType::Struct(vec![Field::new("m", target,
true)].into());
+ let output = cast_update_value(&nested, &nested_target,
CastMode::RowUpdate).unwrap();
+ let map = output
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap()
+ .column(0)
+ .as_any()
+ .downcast_ref::<MapArray>()
+ .unwrap();
+ assert_eq!(
+ map.values().to_data(),
+ Int32Array::from(vec![Some(1), None]).to_data()
+ );
+ assert!(map.is_null(2));
+ }
+
+ #[test]
+ fn map_to_row_constructor_checks_prefix_and_ignores_unused_entries() {
+ use arrow_schema::Field;
+ let entries = StructArray::from(vec![
+ (
+ Arc::new(Field::new("key", DataType::Utf8, false)),
+ Arc::new(StringArray::from(vec![
+ "a", "b", "ignored", "a", "wrong", "b",
+ ])) as ArrayRef,
+ ),
+ (
+ Arc::new(Field::new("value", DataType::Float64, true)),
+ Arc::new(Float64Array::from(vec![1.5, 2.5, f64::NAN, 3.5, 5.0,
6.0])) as ArrayRef,
+ ),
+ ]);
+ let input: ArrayRef = Arc::new(MapArray::new(
+ Arc::new(Field::new("entries", entries.data_type().clone(),
false)),
+ OffsetBuffer::new(vec![0_i32, 3, 4, 6].into()),
+ entries,
+ None,
+ false,
+ ));
+ let target = DataType::Struct(
+ vec![
+ Field::new("a", DataType::Int32, true),
+ Field::new("b", DataType::Int32, true),
+ ]
+ .into(),
+ );
+ for key_type in [
+ DataType::Utf8,
+ DataType::LargeUtf8,
+ DataType::Utf8View,
+ DataType::Binary,
+ DataType::LargeBinary,
+ DataType::BinaryView,
+ ] {
+ let map = input.as_map();
+ let keys = arrow_cast::cast(map.keys().as_ref(),
&key_type).unwrap();
+ let entries = StructArray::from(vec![
+ (Arc::new(Field::new("key", key_type, false)), keys),
+ (
+ Arc::new(Field::new("value", DataType::Float64, true)),
+ map.values().clone(),
+ ),
+ ]);
+ let input: ArrayRef = Arc::new(MapArray::new(
+ Arc::new(Field::new("entries", entries.data_type().clone(),
false)),
+ map.offsets().clone(),
+ entries,
+ None,
+ false,
+ ));
+ assert!(cast_assignment(&input.slice(0, 2), &target).is_err());
+ assert!(cast_update_value(&input, &target,
CastMode::RowUpdate).is_err());
+ let output =
+ cast_update_value(&input.slice(0, 2), &target,
CastMode::RowUpdate).unwrap();
+ let row = output.as_struct();
+ assert_eq!(
+ row.column(0).to_data(),
+ Int32Array::from(vec![1, 3]).to_data()
+ );
+ assert_eq!(
+ row.column(1).to_data(),
+ Int32Array::from(vec![Some(2), None]).to_data()
+ );
+ }
+ }
+
+ #[test]
+ fn list_pairs_constructor_is_top_level_only_and_preserves_slices() {
+ use arrow_array::builder::{Float64Builder, ListBuilder};
+ use arrow_schema::Field;
+ let mut builder =
ListBuilder::new(ListBuilder::new(Float64Builder::new()));
+ for row in [
+ Some(vec![vec![Some(999.0)]]),
+ Some(vec![vec![Some(1.9), Some(10.9)], vec![Some(2.0), None]]),
+ None,
+ Some(vec![]),
+ ] {
+ if let Some(row) = row {
+ for pair in row {
+ for value in pair {
+ builder.values().values().append_option(value);
+ }
+ builder.values().append(true);
+ }
+ builder.append(true);
+ } else {
+ builder.append(false);
+ }
+ }
+ let all: ArrayRef = Arc::new(builder.finish());
+ let input = all.slice(1, 3);
+ let target = test_map_type(DataType::Int32, DataType::Int32);
+ assert!(cast_update_value(&all, &target,
CastMode::RowUpdate).is_err());
+ let output = cast_update_value(&input, &target,
CastMode::RowUpdate).unwrap();
+ let map = output.as_any().downcast_ref::<MapArray>().unwrap();
+ assert_eq!(map.keys().to_data(), Int32Array::from(vec![1,
2]).to_data());
+ assert_eq!(
+ map.values().to_data(),
+ Int32Array::from(vec![Some(10), None]).to_data()
+ );
+ assert!(map.is_null(1));
+ assert_eq!(map.value_length(2), 0);
+ let DataType::List(field) = input.data_type() else {
+ unreachable!()
+ };
+ for layout in [
+ DataType::ListView(field.clone()),
+ DataType::LargeListView(field.clone()),
+ ] {
+ let view = arrow_cast::cast(input.as_ref(), &layout).unwrap();
+ assert_eq!(
+ cast_update_value(&view, &target, CastMode::RowUpdate)
+ .unwrap()
+ .to_data(),
+ output.to_data()
+ );
+ }
+ let nested: ArrayRef = Arc::new(StructArray::from(vec![(
+ Arc::new(Field::new("m", input.data_type().clone(), true)),
+ input,
+ )]));
+ let nested_target = DataType::Struct(vec![Field::new("m", target,
true)].into());
+ assert!(cast_update_value(&nested, &nested_target,
CastMode::RowUpdate).is_err());
+ }
+
+ #[test]
+ fn struct_cast_order_and_nullability_select_whole_column_constructor() {
+ use arrow_schema::Field;
+ let input: ArrayRef = Arc::new(StructArray::from(vec![
+ (
+ Arc::new(Field::new("a", DataType::Int32, true)),
+ Arc::new(Int32Array::from(vec![2])) as ArrayRef,
+ ),
+ (
+ Arc::new(Field::new("b", DataType::Int32, true)),
+ Arc::new(Int32Array::from(vec![1])) as ArrayRef,
+ ),
+ ]));
+ let reversed = DataType::Struct(
+ vec![
+ Field::new("b", DataType::Utf8, true),
+ Field::new("a", DataType::Int32, true),
+ ]
+ .into(),
+ );
+ assert!(cast_assignment(&input, &reversed).is_err());
+ // The reordered constructor cannot inherit the safe cast's
int->string conversion.
+ assert!(cast_update_value(&input, &reversed,
CastMode::RowUpdate).is_err());
+ let required = DataType::Struct(
+ vec![
+ Field::new("a", DataType::Int32, false),
+ Field::new("b", DataType::Int32, true),
+ ]
+ .into(),
+ );
+ assert!(cast_assignment(&input, &required).is_err());
+ assert!(cast_update_value(&input, &required,
CastMode::RowUpdate).is_ok());
+ }
+ #[test]
+ fn dict_map_ambiguity_checks_dictionary_logical_nulls() {
+ use arrow_array::{types::Int32Type, DictionaryArray};
+ use arrow_schema::Field;
+ let child: ArrayRef = Arc::new(
+ DictionaryArray::<Int32Type>::try_new(
+ Int32Array::from(vec![0, 1]),
+ Arc::new(Int32Array::from(vec![Some(1), None])),
+ )
+ .unwrap(),
+ );
+ let input: ArrayRef = Arc::new(StructArray::from(vec![(
+ Arc::new(Field::new("a", child.data_type().clone(), true)),
+ child,
+ )]));
+ let target = test_map_type(DataType::Utf8, DataType::Int32);
+ assert!(cast_update_value(&input, &target,
CastMode::RowUpdate).is_err());
+ let output = cast_update_value(&input.slice(0, 1), &target,
CastMode::RowUpdate).unwrap();
+ assert_eq!(
+ output.as_map().values().to_data(),
+ Int32Array::from(vec![1]).to_data()
+ );
+ }
}
diff --git a/crates/paimon/src/table/upsert_key_matcher.rs
b/crates/paimon/src/table/upsert_key_matcher.rs
index c17de65a..e5257dc4 100644
--- a/crates/paimon/src/table/upsert_key_matcher.rs
+++ b/crates/paimon/src/table/upsert_key_matcher.rs
@@ -63,6 +63,9 @@ pub(super) fn supported_key_type(data_type: &DataType) ->
bool {
| DataType::FixedSizeBinary(_)
| DataType::Date32
| DataType::Date64
+ | DataType::Time32(_)
+ | DataType::Time64(_)
+ | DataType::Timestamp(_, _)
| DataType::Decimal128(_, _)
| DataType::Decimal256(_, _)
)
@@ -243,4 +246,62 @@ mod tests {
.to_string()
.contains("upsert key type differs"));
}
+ #[test]
+ fn temporal_keys_keep_units_nulls_and_composite_identity() {
+ use arrow_schema::TimeUnit;
+ for data_type in [
+ DataType::Time32(TimeUnit::Millisecond),
+ DataType::Timestamp(TimeUnit::Millisecond, None),
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ DataType::Timestamp(TimeUnit::Nanosecond, None),
+ DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
+ DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into())),
+ ] {
+ let temporal = |values: Vec<Option<i64>>| {
+ let array: ArrayRef = if matches!(data_type,
DataType::Time32(_)) {
+ Arc::new(Int32Array::from_iter(
+ values.into_iter().map(|v| v.map(|v| v as i32)),
+ ))
+ } else {
+ Arc::new(Int64Array::from(values))
+ };
+ arrow_cast::cast(array.as_ref(), &data_type).unwrap()
+ };
+ let source = RecordBatch::try_from_iter([
+ (
+ "key",
+ temporal(vec![Some(9), None, Some(10), Some(12), Some(10),
Some(10)]),
+ ),
+ (
+ "part",
+ Arc::new(StringArray::from(vec!["a", "a", "a", "a", "a",
"b"])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let existing = RecordBatch::try_from_iter([
+ (
+ "key",
+ temporal(vec![Some(10), Some(10), None, Some(9), Some(11),
Some(10)]),
+ ),
+ (
+ "part",
+ Arc::new(StringArray::from(vec!["a", "a", "a", "a", "a",
"b"])) as ArrayRef,
+ ),
+ (
+ ROW_ID,
+ Arc::new(Int64Array::from(vec![10, 11, 12, 13, 14, 15]))
as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let mut matcher =
+ UpsertKeyMatcher::new(&source, vec!["key".into(),
"part".into()]).unwrap();
+ matcher.add_existing_batch(&existing.slice(0, 2)).unwrap();
+ matcher.add_existing_batch(&existing.slice(2, 4)).unwrap();
+ assert_eq!(
+ matcher.finish(),
+ (vec![0, 1, 4, 4, 5], vec![13, 12, 10, 11, 15], vec![3]),
+ "{data_type:?}"
+ );
+ }
+ }
}
diff --git a/crates/paimon/src/table/write_batch_normalize.rs
b/crates/paimon/src/table/write_batch_normalize.rs
index 01529d9b..dcd0271a 100644
--- a/crates/paimon/src/table/write_batch_normalize.rs
+++ b/crates/paimon/src/table/write_batch_normalize.rs
@@ -86,13 +86,21 @@ pub(super) fn normalize_write_array(array: &ArrayRef,
expected: &DataType) -> Re
}
(DataType::Map(_, _), DataType::Map(expected_entries,
expected_ordered)) => {
let map = array.as_any().downcast_ref::<MapArray>().unwrap();
- let entries: ArrayRef = Arc::new(map.entries().clone());
- let entries = normalize_write_array(&entries,
expected_entries.data_type())?;
- let entries = entries
- .as_any()
- .downcast_ref::<StructArray>()
- .unwrap()
- .clone();
+ // MAP key/value roles are positional. Their Arrow field names are
+ // aliases, unlike user-defined ROW field names.
+ let DataType::Struct(fields) = expected_entries.data_type() else {
+ return Err(Error::DataInvalid {
+ message: "MAP entries must have a struct type".into(),
+ source: None,
+ });
+ };
+ let columns = [map.keys(), map.values()]
+ .into_iter()
+ .zip(fields)
+ .map(|(column, field)| normalize_write_array(column,
field.data_type()))
+ .collect::<Result<Vec<_>>>()?;
+ let entries =
+ StructArray::try_new(fields.clone(), columns,
None).map_err(normalization_error)?;
let normalized = MapArray::try_new(
expected_entries.clone(),
map.offsets().clone(),
@@ -193,9 +201,9 @@ mod tests {
#[test]
fn map_value_list_alias_is_normalized_recursively() {
- let key_field = Arc::new(Field::new("key", DataType::Int32, false));
+ let key_field = Arc::new(Field::new("source_key", DataType::Int32,
false));
let value_field = Arc::new(Field::new(
- "value",
+ "source_value",
DataType::List(Arc::new(Field::new("item", DataType::Int32,
true))),
true,
));
@@ -229,7 +237,13 @@ mod tests {
));
let expected_entries = Arc::new(Field::new(
"entries",
- DataType::Struct(vec![key_field, expected_value].into()),
+ DataType::Struct(
+ vec![
+ Arc::new(Field::new("key", DataType::Int32, false)),
+ expected_value,
+ ]
+ .into(),
+ ),
false,
));
let expected = DataType::Map(expected_entries, false);
diff --git a/crates/paimon/tests/table_update_nested_test.rs
b/crates/paimon/tests/table_update_nested_test.rs
new file mode 100644
index 00000000..8bf358b4
--- /dev/null
+++ b/crates/paimon/tests/table_update_nested_test.rs
@@ -0,0 +1,439 @@
+// 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.
+
+mod common;
+
+use std::sync::Arc;
+
+use arrow_array::builder::{Int32Builder, ListBuilder};
+use arrow_array::{Array, ArrayRef, Int32Array, ListArray, RecordBatch,
StringArray};
+use arrow_schema::{DataType as ArrowType, Field};
+use futures::TryStreamExt;
+use paimon::spec::{ArrayType, DataType, IntType, Schema, TableSchema};
+use paimon::table::Table;
+
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs, write_batch};
+
+async fn table() -> Table {
+ let path = "memory:/nested_upsert";
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "value",
+ DataType::Array(ArrayType::new(DataType::Int(IntType::new()))),
+ )
+ .column("tag", DataType::Int(IntType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap();
+ let (io, table) = memory_table(path, TableSchema::new(0, &schema));
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ table
+}
+
+fn batch(ids: Vec<i32>, values: Vec<Option<Vec<Option<i32>>>>, tags: Vec<i32>)
-> RecordBatch {
+ let mut builder =
ListBuilder::new(Int32Builder::new()).with_field(Arc::new(Field::new(
+ "item",
+ ArrowType::Int32,
+ true,
+ )));
+ for value in values {
+ if let Some(values) = value {
+ for value in values {
+ builder.values().append_option(value);
+ }
+ builder.append(true);
+ } else {
+ builder.append(false);
+ }
+ }
+ // The upsert source may reorder top-level fields and use Arrow child
aliases.
+ RecordBatch::try_from_iter([
+ ("tag", Arc::new(Int32Array::from(tags)) as ArrayRef),
+ ("value", Arc::new(builder.finish()) as ArrayRef),
+ ("id", Arc::new(Int32Array::from(ids)) as ArrayRef),
+ ])
+ .unwrap()
+}
+
+#[tokio::test]
+async fn nested_upsert_normalizes_before_matching_and_appending() {
+ let table = table().await;
+ let seed = batch(vec![1, 1, 2], vec![None, None, None], vec![10, 11, 12]);
+ write_batch(&table, &seed.project(&[2, 1, 0]).unwrap()).await;
+ let mut update = table.new_write_builder().new_update().unwrap();
+ update.with_update_type(vec!["value".into()]).unwrap();
+ let source = batch(
+ vec![0, 1, 1, 3],
+ vec![
+ Some(vec![Some(999)]),
+ Some(vec![Some(99)]),
+ Some(vec![Some(10), None]),
+ Some(vec![]),
+ ],
+ vec![20, 21, 22, 23],
+ );
+ let messages = update
+ .upsert_by_arrow_with_key(
+ vec![source.slice(1, 1), source.slice(2, 2)],
+ vec!["id".into()],
+ )
+ .await
+ .unwrap();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+ let read = table.new_read_builder();
+ let plan = read.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut rows = Vec::new();
+ for batch in batches {
+ let ids = batch
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let tags = batch
+ .column_by_name("tag")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let values = batch
+ .column_by_name("value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .unwrap();
+ for i in 0..batch.num_rows() {
+ let value = (!values.is_null(i)).then(|| {
+ values
+ .value(i)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>()
+ });
+ rows.push((ids.value(i), tags.value(i), value));
+ }
+ }
+ rows.sort();
+ assert_eq!(
+ rows,
+ vec![
+ (1, 10, Some(vec![Some(10), None])),
+ (1, 11, Some(vec![Some(10), None])),
+ (2, 12, None),
+ (3, 23, Some(vec![]))
+ ]
+ );
+}
+
+#[tokio::test]
+async fn invalid_nested_upsert_batch_creates_no_staged_files() {
+ let table = table().await;
+ let seed = batch(vec![1], vec![None], vec![10]);
+ write_batch(&table, &seed.project(&[2, 1, 0]).unwrap()).await;
+ let paths = || async {
+ let mut files: Vec<_> = table
+ .file_io()
+ .list_status_recursive(table.location())
+ .await
+ .unwrap()
+ .into_iter()
+ .map(|status| status.path)
+ .filter(|path| path.ends_with(".parquet"))
+ .collect();
+ files.sort();
+ files
+ };
+ let before = paths().await;
+ let invalid = RecordBatch::try_from_iter([
+ ("tag", Arc::new(Int32Array::from(vec![20])) as ArrayRef),
+ (
+ "value",
+ Arc::new(StringArray::from(vec!["not an array"])) as ArrayRef,
+ ),
+ ("id", Arc::new(Int32Array::from(vec![2])) as ArrayRef),
+ ])
+ .unwrap();
+ let update = table.new_write_builder().new_update().unwrap();
+ assert!(update
+ .upsert_by_arrow_with_key(vec![seed.clone(), invalid],
vec!["id".into()])
+ .await
+ .is_err());
+ assert_eq!(paths().await, before);
+ let messages = update
+ .upsert_by_arrow_with_key(vec![seed], vec!["id".into()])
+ .await
+ .unwrap();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+}
+
+#[tokio::test]
+async fn temporal_upsert_keys_match_at_declared_precision() {
+ use arrow_array::Int64Array;
+ use paimon::spec::{LocalZonedTimestampType, TimeType, TimestampType};
+ for key_type in [
+ DataType::Time(TimeType::new(3).unwrap()),
+ DataType::Timestamp(TimestampType::new(3).unwrap()),
+ DataType::Timestamp(TimestampType::new(6).unwrap()),
+ DataType::Timestamp(TimestampType::new(9).unwrap()),
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(3).unwrap()),
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(6).unwrap()),
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(9).unwrap()),
+ ] {
+ let start = if matches!(key_type, DataType::Time(_)) {
+ 10
+ } else {
+ -10
+ };
+ let schema = Schema::builder()
+ .column("key", key_type.clone())
+ .column("value", DataType::Int(IntType::new()))
+ .column("tag", DataType::Int(IntType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap();
+ let path = "memory:/temporal_upsert";
+ let (io, table) = memory_table(path, TableSchema::new(0, &schema));
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ let arrow_schema =
+
paimon::arrow::build_target_arrow_schema(table.schema().fields()).unwrap();
+ let batch = |keys: Vec<Option<i64>>, values: Vec<i32>, tags: Vec<i32>|
{
+ let physical: ArrayRef = if matches!(key_type, DataType::Time(_)) {
+ Arc::new(Int32Array::from_iter(
+ keys.into_iter().map(|v| v.map(|v| v as i32)),
+ ))
+ } else {
+ Arc::new(Int64Array::from(keys))
+ };
+ RecordBatch::try_new(
+ arrow_schema.clone(),
+ vec![
+ arrow_cast::cast(physical.as_ref(),
arrow_schema.field(0).data_type()).unwrap(),
+ Arc::new(Int32Array::from(values)),
+ Arc::new(Int32Array::from(tags)),
+ ],
+ )
+ .unwrap()
+ };
+ write_batch(
+ &table,
+ &batch(
+ vec![Some(start), Some(start), None, Some(start + 1)],
+ vec![10; 4],
+ vec![0, 1, 2, 3],
+ ),
+ )
+ .await;
+ let mut update = table.new_write_builder().new_update().unwrap();
+ update.with_update_type(vec!["value".into()]).unwrap();
+ let source = batch(
+ vec![None, Some(start), Some(start + 2), Some(start)],
+ vec![20, 30, 40, 50],
+ vec![99, 99, 4, 99],
+ );
+ let messages = update
+ .upsert_by_arrow_with_key(
+ vec![source.slice(0, 2), source.slice(2, 2)],
+ vec!["key".into()],
+ )
+ .await
+ .unwrap();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+ let read = table.new_read_builder();
+ let plan = read.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut actual = Vec::new();
+ for batch in batches {
+ let keys = arrow_cast::cast(batch.column(0).as_ref(),
&ArrowType::Int64).unwrap();
+ let keys = keys.as_any().downcast_ref::<Int64Array>().unwrap();
+ let values = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let tags = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ for row in 0..batch.num_rows() {
+ actual.push((
+ tags.value(row),
+ (!keys.is_null(row)).then(|| keys.value(row)),
+ values.value(row),
+ ));
+ }
+ }
+ actual.sort();
+ assert_eq!(
+ actual,
+ vec![
+ (0, Some(start), 50),
+ (1, Some(start), 50),
+ (2, None, 20),
+ (3, Some(start + 1), 10),
+ (4, Some(start + 2), 40)
+ ],
+ "{key_type:?}"
+ );
+ }
+}
+
+#[tokio::test]
+async fn upsert_does_not_read_unrelated_composite_partitions() {
+ use paimon::spec::VarCharType;
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("pt", DataType::VarChar(VarCharType::new(20).unwrap()))
+ .column("sub", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .partition_keys(["pt", "sub"])
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap();
+ let path = "memory:/partition_upsert";
+ let (io, table) = memory_table(path, TableSchema::new(0, &schema));
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ let arrow_schema =
paimon::arrow::build_target_arrow_schema(table.schema().fields()).unwrap();
+ let batch = |pt: Vec<Option<&str>>, sub: Vec<i32>, values: Vec<i32>| {
+ RecordBatch::try_new(
+ arrow_schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1; pt.len()])),
+ Arc::new(StringArray::from(pt)),
+ Arc::new(Int32Array::from(sub)),
+ Arc::new(Int32Array::from(values)),
+ ],
+ )
+ .unwrap()
+ };
+ write_batch(
+ &table,
+ &batch(
+ vec![Some("a"), Some("a"), Some("b"), None, None],
+ vec![1, 2, 1, 1, 2],
+ vec![10; 5],
+ ),
+ )
+ .await;
+ let scan = table.new_read_builder().new_scan().plan().await.unwrap();
+ let relevant = |path: &str| {
+ path.contains("/pt=a/sub=1/")
+ || path.contains("/pt=b/sub=2/")
+ || path.contains("/pt=__DEFAULT_PARTITION__/sub=1/")
+ };
+ let mut removed = 0;
+ for split in scan.splits() {
+ if !relevant(split.bucket_path()) {
+ for file in split.data_files() {
+ io.delete_file(&format!("{}/{}", split.bucket_path(),
file.file_name))
+ .await
+ .unwrap();
+ removed += 1;
+ }
+ }
+ }
+ assert_eq!(removed, 3);
+ // Implicit partition matching must use exact tuples, including NULL. A
+ // Cartesian product would still read a/sub=2, b/sub=1 and NULL/sub=2.
+ let source = batch(
+ vec![Some("a"), Some("b"), None, Some("a")],
+ vec![1, 2, 1, 1],
+ vec![15, 30, 40, 20],
+ );
+ let mut update = table.new_write_builder().new_update().unwrap();
+ update.with_update_type(vec!["value".into()]).unwrap();
+ let messages = update
+ .upsert_by_arrow_with_key(vec![source], vec!["id".into()])
+ .await
+ .unwrap();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+ let read = table.new_read_builder();
+ let plan = read.new_scan().plan().await.unwrap();
+ let splits: Vec<_> = plan
+ .splits()
+ .iter()
+ .filter(|split| relevant(split.bucket_path()))
+ .cloned()
+ .collect();
+ let batches: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(&splits)
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut values: Vec<_> = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column_by_name("value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect();
+ values.sort();
+ assert_eq!(values, vec![20, 30, 40]);
+}
