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 2825759f fix: repair native vector writes and data evolution read
ranges (#1090)
2825759f is described below
commit 2825759f5894534a7cf041498bc3c3914f448ec6
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 10 15:04:58 2026 +0800
fix: repair native vector writes and data evolution read ranges (#1090)
---
crates/paimon/src/arrow/format/mod.rs | 31 ++++++
crates/paimon/src/table/data_evolution_reader.rs | 113 +++++++++++++++----
crates/paimon/src/table/table_write.rs | 23 ++++
crates/paimon/src/table/write_batch_normalize.rs | 134 ++++++++++++++++++++++-
4 files changed, 276 insertions(+), 25 deletions(-)
diff --git a/crates/paimon/src/arrow/format/mod.rs
b/crates/paimon/src/arrow/format/mod.rs
index dc42feaf..3f12a841 100644
--- a/crates/paimon/src/arrow/format/mod.rs
+++ b/crates/paimon/src/arrow/format/mod.rs
@@ -476,6 +476,21 @@ fn supported_write_formats() -> Vec<&'static str> {
]
}
+/// Reject unavailable factories before a table writer accepts any data.
+pub(crate) fn validate_write_format(format: &str) -> crate::Result<()> {
+ let extension = format!(".{}", format.to_ascii_lowercase());
+ let supported = supported_write_formats();
+ if supported.contains(&extension.as_str()) {
+ return Ok(());
+ }
+ Err(Error::Unsupported {
+ message: format!(
+ "unsupported write format '{format}': expected {}",
+ supported.join(", ")
+ ),
+ })
+}
+
/// Reusable factory for one rolling writer, mirroring Java
FormatWriterFactory.
/// Output and compression belong to each file; schema/options belong to the
factory.
#[async_trait]
@@ -712,6 +727,22 @@ mod tests {
use crate::io::FileIOBuilder;
use crate::spec::{DataType, IntType};
+ #[test]
+ fn write_format_validation_matches_compiled_factories() {
+ for extension in supported_write_formats() {
+ validate_write_format(extension.trim_start_matches('.')).unwrap();
+ }
+ validate_write_format("PARQUET").unwrap();
+ assert!(matches!(
+ validate_write_format("lance"),
+ Err(Error::Unsupported { .. })
+ ));
+ assert_eq!(
+ validate_write_format("vortex").is_ok(),
+ cfg!(feature = "vortex")
+ );
+ }
+
#[test]
fn format_selects_physical_or_projected_fields() {
let data_schema_fields = vec![
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index b856bbc8..b1dbbddb 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -1824,17 +1824,9 @@ fn open_source_stream(
FieldSource::VectorBunch {
bunch, data_fields, ..
} => {
- let anchor =
crate::table::source::data_evolution_anchor_file(split.data_files())?;
- let first_row_id = anchor.first_row_id.ok_or_else(||
Error::DataInvalid {
- message: format!(
- "Data-evolution anchor file '{}' is missing first_row_id",
- anchor.file_name
- ),
- source: None,
- })?;
let selected_ranges = selected_absolute_row_ranges_for_file(
- first_row_id,
- anchor.row_count,
+ bunch.expected_first_row_id,
+ bunch.expected_row_count,
row_ranges.as_deref(),
anchor_deletion_vector.map(|context|
context.deletion_vector.as_ref()),
)?;
@@ -2469,6 +2461,7 @@ fn build_source_plan_with_row_id_pushdown(
let source_idx = sources.len();
sources.push(FieldSource::VectorBunch {
bunch: VectorBunch::new(
+ prepared_group.first_row_id,
prepared_group.logical_row_count,
file.schema_id,
format_suffix,
@@ -3015,6 +3008,7 @@ struct VectorBunch {
schema_id: i64,
format_suffix: String,
normalized_write_cols: Vec<String>,
+ expected_first_row_id: i64,
expected_row_count: i64,
latest_first_row_id: i64,
expected_next_first_row_id: i64,
@@ -3025,6 +3019,7 @@ struct VectorBunch {
impl VectorBunch {
fn new(
+ expected_first_row_id: i64,
expected_row_count: i64,
schema_id: i64,
format_suffix: String,
@@ -3035,6 +3030,7 @@ impl VectorBunch {
schema_id,
format_suffix,
normalized_write_cols,
+ expected_first_row_id,
expected_row_count,
latest_first_row_id: -1,
expected_next_first_row_id: -1,
@@ -3959,7 +3955,7 @@ mod tests {
#[test]
fn test_vector_bunch_aggregates_contiguous_segments() {
- let mut bunch = VectorBunch::new(30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
+ let mut bunch = VectorBunch::new(0, 30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
bunch
.add(
data_file("v1.vector.parquet", 0, 10, 1, Some(vec!["emb"])),
@@ -3992,7 +3988,7 @@ mod tests {
#[test]
fn test_vector_bunch_rejects_gap() {
- let mut bunch = VectorBunch::new(30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
+ let mut bunch = VectorBunch::new(0, 30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
bunch
.add(
data_file("v1.vector.parquet", 0, 10, 1, Some(vec!["emb"])),
@@ -4013,7 +4009,7 @@ mod tests {
#[test]
fn test_vector_bunch_ignores_same_first_row_id_lower_seq() {
- let mut bunch = VectorBunch::new(30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
+ let mut bunch = VectorBunch::new(0, 30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
bunch
.add(
data_file("v-high.vector.parquet", 0, 10, 3,
Some(vec!["emb"])),
@@ -4034,7 +4030,7 @@ mod tests {
#[test]
fn test_vector_bunch_rejects_same_first_row_id_higher_seq() {
- let mut bunch = VectorBunch::new(30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
+ let mut bunch = VectorBunch::new(0, 30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
bunch
.add(
data_file("v-low.vector.parquet", 0, 10, 2, Some(vec!["emb"])),
@@ -4052,7 +4048,7 @@ mod tests {
#[test]
fn test_vector_bunch_ignores_overlapping_lower_seq() {
- let mut bunch = VectorBunch::new(30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
+ let mut bunch = VectorBunch::new(0, 30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
bunch
.add(
data_file("v1.vector.parquet", 0, 10, 3, Some(vec!["emb"])),
@@ -4072,7 +4068,7 @@ mod tests {
#[test]
fn test_selected_vector_bunch_rejects_partial_higher_sequence_overlap() {
- let mut bunch = VectorBunch::new(30, 0, "parquet".to_string(),
vec!["emb".to_string()])
+ let mut bunch = VectorBunch::new(0, 30, 0, "parquet".to_string(),
vec!["emb".to_string()])
.with_row_id_pushdown(true);
bunch
.add(
@@ -4097,7 +4093,7 @@ mod tests {
#[test]
fn test_vector_bunch_rejects_row_count_overflow() {
- let mut bunch = VectorBunch::new(15, 0, "parquet".to_string(),
vec!["emb".to_string()]);
+ let mut bunch = VectorBunch::new(0, 15, 0, "parquet".to_string(),
vec!["emb".to_string()]);
bunch
.add(
data_file("v1.vector.parquet", 0, 10, 1, Some(vec!["emb"])),
@@ -4116,7 +4112,7 @@ mod tests {
#[test]
fn test_vector_bunch_rejects_key_identity_mismatch() {
// schema_id mismatch
- let mut bunch = VectorBunch::new(30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
+ let mut bunch = VectorBunch::new(0, 30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
bunch
.add(
data_file("v1.vector.parquet", 0, 10, 1, Some(vec!["emb"])),
@@ -4129,7 +4125,7 @@ mod tests {
assert!(matches!(err, Error::DataInvalid { .. }));
// format_suffix mismatch
- let mut bunch2 = VectorBunch::new(30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
+ let mut bunch2 = VectorBunch::new(0, 30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
bunch2
.add(
data_file("v1.vector.parquet", 0, 10, 1, Some(vec!["emb"])),
@@ -4145,7 +4141,7 @@ mod tests {
assert!(matches!(err2, Error::DataInvalid { .. }));
// normalized_write_cols mismatch
- let mut bunch3 = VectorBunch::new(30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
+ let mut bunch3 = VectorBunch::new(0, 30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
bunch3
.add(
data_file("v1.vector.parquet", 0, 10, 1, Some(vec!["emb"])),
@@ -4163,7 +4159,7 @@ mod tests {
#[test]
fn test_vector_bunch_rejects_non_vector_file() {
- let mut bunch = VectorBunch::new(30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
+ let mut bunch = VectorBunch::new(0, 30, 0, "parquet".to_string(),
vec!["emb".to_string()]);
let err = bunch
.add(
data_file("v1.parquet", 0, 10, 1, Some(vec!["emb"])),
@@ -6093,6 +6089,81 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_filtered_vector_read_keeps_each_merge_group_range() {
+ let tempdir = tempdir().unwrap();
+ let bucket_dir = tempdir.path().join("bucket-0");
+ fs::create_dir_all(&bucket_dir).unwrap();
+ let mut files = Vec::new();
+ for (index, first_row_id) in [100, 200].into_iter().enumerate() {
+ let ids = vec![index as i32 * 2 + 1, index as i32 * 2 + 2];
+ let normal_name = format!("data-{index}.parquet");
+ let vector_name = format!("data-{index}.vector.parquet");
+ let normal_path = bucket_dir.join(&normal_name);
+ let vector_path = bucket_dir.join(&vector_name);
+ write_int_parquet_file(&normal_path, vec![("id", ids)], None);
+ let vector = vec![index as f32 + 1.0, index as f32 + 2.0];
+ write_fixed_size_list_parquet(
+ &vector_path,
+ "embedding",
+ 2,
+ &[Some(vector.clone()), Some(vector)],
+ );
+ for (name, path, columns) in [
+ (normal_name, normal_path, vec!["id"]),
+ (vector_name, vector_path, vec!["embedding"]),
+ ] {
+ files.push(data_file_meta_with_path(
+ &name,
+ first_row_id,
+ 2,
+ 1,
+ path.metadata().unwrap().len() as i64,
+ Some(columns),
+ ));
+ }
+ }
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("embedding", vector_float_type(2))
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ FileIOBuilder::new("file").build().unwrap(),
+ Identifier::new("default", "vector_groups"),
+ local_file_path(tempdir.path()),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ // A Python plan can bundle multiple independent file groups in one
+ // split. Each group's vectors must use its own normal-file anchor.
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(local_file_path(&bucket_dir))
+ .with_total_buckets(1)
+ .with_data_files(files)
+ .with_row_ranges(vec![RowRange::new(100, 100), RowRange::new(201,
201)])
+ .build()
+ .unwrap();
+ let batches = TableRead::new(&table, table.schema().fields().to_vec(),
Vec::new())
+ .to_arrow(&[split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(collect_int_values(&batches, "id"), vec![1, 4]);
+ assert_fixed_size_list(
+ &batches,
+ "embedding",
+ 2,
+ &[Some(vec![1.0, 2.0]), Some(vec![2.0, 3.0])],
+ );
+ }
+
/// (2) Same shape but the dedicated vector file is `.vector.vortex`.
#[cfg(feature = "vortex")]
#[tokio::test]
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index f4f055a8..5c153f25 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -486,6 +486,12 @@ impl TableWrite {
.fields()
.iter()
.any(|f| matches!(f.data_type(), DataType::Vector(_)));
+ if let Some(format) = vector_file_format
+ .as_deref()
+ .filter(|_| has_dedicated_vector_fields)
+ {
+ crate::arrow::format::validate_write_format(format)?;
+ }
let file_index_options = FileIndexOptions::parse(schema.options(),
schema.fields())?;
if file_index_options.is_some() && has_dedicated_vector_fields {
@@ -3507,6 +3513,23 @@ pub(in crate::table) mod tests {
.await;
}
+ #[test]
+ fn test_unsupported_vector_format_fails_before_accepting_data() {
+ let file_io = FileIO::from_path("memory:/").unwrap().build().unwrap();
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "unsupported_vector_format"),
+ "memory:/unsupported_vector_format".to_string(),
+ test_vector_table_schema("lance"),
+ None,
+ );
+ let error = TableWrite::new(&table, "test-user".to_string())
+ .err()
+ .unwrap();
+ assert!(matches!(error, crate::Error::Unsupported { .. }));
+ assert!(error.to_string().contains("lance"));
+ }
+
#[cfg(feature = "vortex")]
#[tokio::test]
async fn test_vector_write_uses_dedicated_vortex_file() {
diff --git a/crates/paimon/src/table/write_batch_normalize.rs
b/crates/paimon/src/table/write_batch_normalize.rs
index 0156133c..57fce025 100644
--- a/crates/paimon/src/table/write_batch_normalize.rs
+++ b/crates/paimon/src/table/write_batch_normalize.rs
@@ -25,8 +25,8 @@
use crate::{Error, Result};
use arrow_array::{
- Array, ArrayRef, BinaryArray, Decimal128Array, FixedSizeBinaryArray,
ListArray, MapArray,
- StringArray, StructArray,
+ Array, ArrayRef, BinaryArray, Decimal128Array, FixedSizeBinaryArray,
FixedSizeListArray,
+ ListArray, MapArray, StringArray, StructArray,
};
use arrow_buffer::{NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow_schema::{DataType, Field};
@@ -218,7 +218,10 @@ fn normalize_visible_array(
if array.data_type() == expected
&& !matches!(
expected,
- DataType::List(_) | DataType::Struct(_) | DataType::Map(_, _)
+ DataType::List(_)
+ | DataType::FixedSizeList(_, _)
+ | DataType::Struct(_)
+ | DataType::Map(_, _)
)
{
return Ok(array.clone());
@@ -244,6 +247,26 @@ fn normalize_visible_array(
.map_err(normalization_error)?;
Ok(Arc::new(normalized))
}
+ (
+ DataType::FixedSizeList(_, actual_size),
+ DataType::FixedSizeList(expected_child, expected_size),
+ ) if actual_size == expected_size => {
+ let list =
array.as_any().downcast_ref::<FixedSizeListArray>().unwrap();
+ // VECTOR elements are positional, just like ARRAY elements. Reuse
+ // the values of sliced inputs and let Arrow validate NOT NULL
+ // elements against the combined vector/containing ROW validity.
+ let nulls = NullBuffer::union(list.nulls(), parent_nulls);
+ let values = normalize_write_array(list.values(),
expected_child.data_type())?;
+ let normalized = FixedSizeListArray::try_new_with_length(
+ expected_child.clone(),
+ *expected_size,
+ values,
+ nulls,
+ list.len(),
+ )
+ .map_err(normalization_error)?;
+ Ok(Arc::new(normalized))
+ }
(DataType::Struct(actual_fields), DataType::Struct(expected_fields))
if actual_fields.len() == expected_fields.len() =>
{
@@ -392,10 +415,113 @@ fn normalization_error(error: arrow_schema::ArrowError)
-> Error {
#[cfg(test)]
mod tests {
use super::*;
- use arrow_array::{Int32Array, LargeBinaryArray, StringArray};
+ use arrow_array::{FixedSizeListArray, Int32Array, LargeBinaryArray,
StringArray};
use arrow_buffer::{OffsetBuffer, ScalarBuffer};
use arrow_schema::Field;
+ #[test]
+ fn vector_child_aliases_preserve_sliced_values_and_nulls() {
+ for name in ["item", "", "element"] {
+ let input = FixedSizeListArray::new(
+ Arc::new(Field::new(name, DataType::Int32, true)),
+ 2,
+ Arc::new(Int32Array::from(vec![
+ Some(0),
+ Some(1),
+ Some(2),
+ Some(3),
+ None,
+ None,
+ Some(6),
+ Some(7),
+ ])),
+ Some(NullBuffer::from(vec![true, true, false, true])),
+ )
+ .slice(1, 3);
+ let input: ArrayRef = Arc::new(input);
+ let expected =
+ DataType::FixedSizeList(Arc::new(Field::new("element",
DataType::Int32, true)), 2);
+ let output = normalize_write_array(&input, &expected).unwrap();
+ output.to_data().validate_full().unwrap();
+ assert_eq!(output.data_type(), &expected);
+ assert_eq!(output.len(), 3);
+ assert_eq!(output.nulls(), input.nulls());
+ let source =
input.as_any().downcast_ref::<FixedSizeListArray>().unwrap();
+ let result = output
+ .as_any()
+ .downcast_ref::<FixedSizeListArray>()
+ .unwrap();
+ assert!(Arc::ptr_eq(result.values(), source.values()));
+ assert_eq!(
+ result.value(0).to_data(),
+ Int32Array::from(vec![2, 3]).to_data()
+ );
+ assert_eq!(
+ result.value(2).to_data(),
+ Int32Array::from(vec![6, 7]).to_data()
+ );
+ }
+ }
+
+ #[test]
+ fn vector_normalization_rejects_dimensions_and_element_type_changes() {
+ let input: ArrayRef = Arc::new(FixedSizeListArray::new(
+ Arc::new(Field::new("item", DataType::Int32, true)),
+ 2,
+ Arc::new(Int32Array::from(vec![1, 2])),
+ None,
+ ));
+ for expected in [
+ DataType::FixedSizeList(Arc::new(Field::new("element",
DataType::Int32, true)), 1),
+ DataType::FixedSizeList(Arc::new(Field::new("element",
DataType::Int64, true)), 2),
+ DataType::List(Arc::new(Field::new("element", DataType::Int32,
true))),
+ ] {
+ assert!(normalize_write_array(&input, &expected).is_err());
+ }
+ }
+
+ #[test]
+ fn vector_element_nullability_only_checks_visible_parents() {
+ let expected =
+ DataType::FixedSizeList(Arc::new(Field::new("element",
DataType::Int32, false)), 2);
+ let make_input = |nulls| {
+ Arc::new(FixedSizeListArray::new(
+ Arc::new(Field::new("item", DataType::Int32, true)),
+ 2,
+ Arc::new(Int32Array::from(vec![None, None, Some(3), Some(4)])),
+ nulls,
+ )) as ArrayRef
+ };
+ let hidden = make_input(Some(NullBuffer::from(vec![false, true])));
+ let output = normalize_write_array(&hidden, &expected).unwrap();
+ assert!(output.is_null(0));
+ assert!(!output.is_null(1));
+ assert!(normalize_write_array(&make_input(None), &expected).is_err());
+
+ let row: ArrayRef = Arc::new(StructArray::new(
+ vec![Field::new("embedding", hidden.data_type().clone(),
true)].into(),
+ vec![make_input(None)],
+ Some(NullBuffer::from(vec![false, true])),
+ ));
+ let row_type =
+ DataType::Struct(vec![Field::new("embedding", expected.clone(),
true)].into());
+ let output = normalize_write_array(&row, &row_type).unwrap();
+ let result = output.as_any().downcast_ref::<StructArray>().unwrap();
+ assert!(result.column(0).is_null(0));
+ assert!(!result.column(0).is_null(1));
+
+ // FFI inputs can claim NOT NULL elements while carrying visible NULLs.
+ let invalid = unsafe {
+ make_input(None)
+ .to_data()
+ .into_builder()
+ .data_type(expected.clone())
+ .build_unchecked()
+ };
+ let invalid = arrow_array::make_array(invalid);
+ assert!(normalize_write_array(&invalid, &expected).is_err());
+ }
+
fn blob_map_field(nullable_value: bool) -> Field {
Field::new(
"payload",
