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 976e408 [arrow] Read inline VECTOR columns via Arrow FixedSizeList
(#412)
976e408 is described below
commit 976e4082a06611944d46af85bec4f1603b69c5c1
Author: Junrui Lee <[email protected]>
AuthorDate: Mon Jun 29 19:21:44 2026 +0800
[arrow] Read inline VECTOR columns via Arrow FixedSizeList (#412)
---
crates/paimon/src/arrow/format/mod.rs | 3 +
crates/paimon/src/arrow/format/parquet.rs | 61 ++++++++++
crates/paimon/src/arrow/mod.rs | 108 +++++++++++++++++-
crates/paimon/src/spec/types.rs | 8 +-
crates/paimon/src/table/data_file_reader.rs | 165 ++++++++++++++++++++++++++++
5 files changed, 337 insertions(+), 8 deletions(-)
diff --git a/crates/paimon/src/arrow/format/mod.rs
b/crates/paimon/src/arrow/format/mod.rs
index 1593f5a..cec1be9 100644
--- a/crates/paimon/src/arrow/format/mod.rs
+++ b/crates/paimon/src/arrow/format/mod.rs
@@ -25,6 +25,9 @@ mod row;
#[cfg(feature = "vortex")]
mod vortex;
+#[cfg(test)]
+pub(crate) use parquet::ParquetFormatWriter;
+
use crate::io::{FileRead, OutputFile};
use crate::spec::{DataField, Predicate};
use crate::table::{ArrowRecordBatchStream, RowRange};
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index bec9466..d7bfc5b 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -1410,4 +1410,65 @@ mod tests {
let total_rows: usize = reader.into_iter().map(|r|
r.unwrap().num_rows()).sum();
assert_eq!(total_rows, 5);
}
+
+ #[tokio::test]
+ async fn test_parquet_inline_fixed_size_list_roundtrip() {
+ use arrow_array::builder::{FixedSizeListBuilder, Float32Builder};
+ use arrow_array::{Array, FixedSizeListArray, Float32Array};
+
+ // Build a FixedSizeList<Float32, 2> column: row 0 = [1.0, 2.0], row 1
= null.
+ let mut builder = FixedSizeListBuilder::new(Float32Builder::new(),
2).with_field(Arc::new(
+ ArrowField::new("element", ArrowDataType::Float32, true),
+ ));
+ builder.values().append_value(1.0);
+ builder.values().append_value(2.0);
+ builder.append(true);
+ builder.values().append_value(0.0);
+ builder.values().append_value(0.0);
+ builder.append(false); // null vector row
+ let vec_array = builder.finish();
+
+ let schema = Arc::new(ArrowSchema::new(vec![ArrowField::new(
+ "embedding",
+ ArrowDataType::FixedSizeList(
+ Arc::new(ArrowField::new("element", ArrowDataType::Float32,
true)),
+ 2,
+ ),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(schema.clone(),
vec![Arc::new(vec_array)]).unwrap();
+
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let path = "memory:/test_parquet_inline_vector.parquet";
+ let output = file_io.new_output(path).unwrap();
+ let mut writer: Box<dyn FormatFileWriter> = Box::new(
+ ParquetFormatWriter::new(&output, schema.clone(), "zstd", 1)
+ .await
+ .unwrap(),
+ );
+ writer.write(&batch).await.unwrap();
+ writer.close().await.unwrap();
+
+ let bytes = file_io.new_input(path).unwrap().read().await.unwrap();
+ let reader =
+
parquet::arrow::arrow_reader::ParquetRecordBatchReader::try_new(bytes,
1024).unwrap();
+ let batches: Vec<RecordBatch> = reader.into_iter().map(|r|
r.unwrap()).collect();
+ assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 2);
+
+ let col = batches[0].column(0);
+ let fsl = col
+ .as_any()
+ .downcast_ref::<FixedSizeListArray>()
+ .expect("column should be FixedSizeListArray");
+ assert_eq!(fsl.value_length(), 2);
+ assert!(fsl.is_valid(0));
+ assert!(fsl.is_null(1)); // null vector row preserved
+
+ let row0 = fsl.value(0);
+ let floats = row0
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .expect("child should be Float32Array");
+ assert_eq!(floats.values(), &[1.0, 2.0]);
+ }
}
diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs
index fc6451f..b897c8c 100644
--- a/crates/paimon/src/arrow/mod.rs
+++ b/crates/paimon/src/arrow/mod.rs
@@ -22,7 +22,7 @@ pub(crate) mod schema_evolution;
use crate::spec::{
ArrayType, BigIntType, BooleanType, DataField, DataType as PaimonDataType,
DateType,
DecimalType, DoubleType, FloatType, IntType, LocalZonedTimestampType,
MapType, RowType,
- SmallIntType, TimeType, TimestampType, TinyIntType, VarBinaryType,
VarCharType,
+ SmallIntType, TimeType, TimestampType, TinyIntType, VarBinaryType,
VarCharType, VectorType,
};
use arrow_schema::DataType as ArrowDataType;
use arrow_schema::{Field as ArrowField, Schema as ArrowSchema, TimeUnit};
@@ -117,10 +117,19 @@ pub fn paimon_type_to_arrow(dt: &PaimonDataType) ->
crate::Result<ArrowDataType>
.collect::<crate::Result<Vec<_>>>()?;
ArrowDataType::Struct(fields.into())
}
- PaimonDataType::Vector(_) => {
- return Err(crate::Error::Unsupported {
- message: "VectorType is not yet supported in arrow
conversion".to_string(),
- })
+ PaimonDataType::Vector(v) => {
+ let element_type = paimon_type_to_arrow(v.element_type())?;
+ // VectorType::MAX_LENGTH is i32::MAX as u32 (validated at
construction),
+ // so the length always fits in the i32 Arrow FixedSizeList size.
+ let length = v.length() as i32;
+ ArrowDataType::FixedSizeList(
+ Arc::new(ArrowField::new(
+ "element",
+ element_type,
+ v.element_type().is_nullable(),
+ )),
+ length,
+ )
}
})
}
@@ -226,6 +235,17 @@ pub fn arrow_to_paimon_type(
paimon_fields,
)))
}
+ ArrowDataType::FixedSizeList(field, size) => {
+ let element = arrow_to_paimon_type(field.data_type(),
field.is_nullable())?;
+ // FixedSizeList size is i32; reject non-positive sizes with a
clear error
+ // rather than casting a negative into a huge u32.
+ let length = u32::try_from(*size).map_err(|_|
crate::Error::DataTypeInvalid {
+ message: format!("Invalid vector (FixedSizeList) length:
{size}"),
+ })?;
+ Ok(PaimonDataType::Vector(VectorType::try_new(
+ nullable, length, element,
+ )?))
+ }
_ => Err(crate::Error::Unsupported {
message: format!("Unsupported Arrow type for Paimon conversion:
{arrow_type:?}"),
}),
@@ -465,6 +485,84 @@ mod tests {
assert!(result.is_err());
}
+ #[test]
+ fn test_vector_to_arrow_nullable() {
+ let paimon = PaimonDataType::Vector(
+ VectorType::try_new(true, 128,
PaimonDataType::Float(FloatType::new())).unwrap(),
+ );
+ let expected = ArrowDataType::FixedSizeList(
+ Arc::new(ArrowField::new("element", ArrowDataType::Float32, true)),
+ 128,
+ );
+ assert_paimon_to_arrow(&paimon, &expected);
+ }
+
+ #[test]
+ fn test_vector_to_arrow_not_null_vector_has_float64_child() {
+ // The vector's own non-nullability is not represented in the
ArrowDataType;
+ // only the child element type and length are.
+ let paimon = PaimonDataType::Vector(
+ VectorType::try_new(false, 2,
PaimonDataType::Double(DoubleType::new())).unwrap(),
+ );
+ let arrow = paimon_type_to_arrow(&paimon).unwrap();
+ match arrow {
+ ArrowDataType::FixedSizeList(field, size) => {
+ assert_eq!(size, 2);
+ assert_eq!(field.data_type(), &ArrowDataType::Float64);
+ }
+ other => panic!("expected FixedSizeList, got {other:?}"),
+ }
+ }
+
+ #[test]
+ fn test_arrow_fixed_size_list_to_vector() {
+ let arrow = ArrowDataType::FixedSizeList(
+ Arc::new(ArrowField::new("element", ArrowDataType::Float32, true)),
+ 4,
+ );
+ let paimon = arrow_to_paimon_type(&arrow, true).unwrap();
+ match paimon {
+ PaimonDataType::Vector(v) => {
+ assert_eq!(v.length(), 4);
+ assert_eq!(v.element_type(),
&PaimonDataType::Float(FloatType::new()));
+ }
+ other => panic!("expected Vector, got {other:?}"),
+ }
+ }
+
+ #[test]
+ fn test_arrow_fixed_size_list_invalid_element_rejected() {
+ let arrow = ArrowDataType::FixedSizeList(
+ Arc::new(ArrowField::new("element", ArrowDataType::Utf8, true)),
+ 4,
+ );
+ let err = arrow_to_paimon_type(&arrow, true);
+ assert!(matches!(err, Err(crate::Error::DataTypeInvalid { .. })));
+ }
+
+ #[test]
+ fn test_arrow_fixed_size_list_zero_length_rejected() {
+ let arrow = ArrowDataType::FixedSizeList(
+ Arc::new(ArrowField::new("element", ArrowDataType::Float32, true)),
+ 0,
+ );
+ let err = arrow_to_paimon_type(&arrow, true);
+ assert!(matches!(err, Err(crate::Error::DataTypeInvalid { .. })));
+ }
+
+ #[test]
+ fn test_arrow_fixed_size_list_negative_length_rejected() {
+ // A negative FixedSizeList size IS directly constructible in Arrow,
so it
+ // exercises the `u32::try_from(*size)` negative branch in the
conversion
+ // (distinct from the zero case, which `VectorType::try_new` rejects).
+ let arrow = ArrowDataType::FixedSizeList(
+ Arc::new(ArrowField::new("element", ArrowDataType::Float32, true)),
+ -1,
+ );
+ let err = arrow_to_paimon_type(&arrow, true);
+ assert!(matches!(err, Err(crate::Error::DataTypeInvalid { .. })));
+ }
+
#[test]
fn test_arrow_fields_to_paimon_ids() {
let fields = vec![
diff --git a/crates/paimon/src/spec/types.rs b/crates/paimon/src/spec/types.rs
index 4dbe8e3..7269aff 100644
--- a/crates/paimon/src/spec/types.rs
+++ b/crates/paimon/src/spec/types.rs
@@ -2623,11 +2623,13 @@ mod tests {
}
#[test]
- fn test_datatype_vector_arrow_unsupported() {
+ fn test_datatype_vector_arrow_conversion() {
+ // PR 2 lifted the PR 1 boundary: Vector now converts to an Arrow
+ // FixedSizeList (see arrow/mod.rs for the full conversion test
matrix).
let dt = DataType::Vector(
VectorType::try_new(true, 4,
DataType::Float(FloatType::new())).unwrap(),
);
- let err = crate::arrow::paimon_type_to_arrow(&dt);
- assert!(matches!(err, Err(crate::Error::Unsupported { .. })));
+ let arrow = crate::arrow::paimon_type_to_arrow(&dt).unwrap();
+ assert!(matches!(arrow, arrow_schema::DataType::FixedSizeList(_, 4)));
}
}
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index a780446..111e23e 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -1066,3 +1066,168 @@ mod tests {
assert_eq!(collect_ids(&batches), vec![11]);
}
}
+
+/// Parquet-only end-to-end tests for the inline VECTOR (`FixedSizeList`) read
path.
+///
+/// This module is deliberately NOT gated behind the `mosaic` feature: the
vector
+/// read capability is core parquet support, so these tests must run under a
plain
+/// `cargo test -p paimon`.
+#[cfg(test)]
+mod vector_parquet_tests {
+ use super::*;
+ use crate::arrow::format::FormatFileWriter;
+ use crate::arrow::format::ParquetFormatWriter;
+ use crate::io::FileIOBuilder;
+ use crate::spec::stats::BinaryTableStats;
+ use crate::spec::{DataFileMeta, DataType, FloatType, VectorType};
+ use crate::table::source::DataSplitBuilder;
+ use arrow_array::builder::{FixedSizeListBuilder, Float32Builder};
+ use arrow_array::{FixedSizeListArray, Float32Array, RecordBatch};
+ use arrow_schema::{DataType as ArrowDataType, Field as ArrowField};
+ use futures::TryStreamExt;
+
+ fn data_file(file_name: &str, file_size: i64, row_count: i64, schema_id:
i64) -> DataFileMeta {
+ DataFileMeta {
+ file_name: file_name.to_string(),
+ file_size,
+ row_count,
+ min_key: Vec::new(),
+ max_key: Vec::new(),
+ key_stats: BinaryTableStats::empty(),
+ value_stats: BinaryTableStats::empty(),
+ min_sequence_number: 0,
+ max_sequence_number: 0,
+ schema_id,
+ level: 0,
+ extra_files: Vec::new(),
+ creation_time: None,
+ delete_row_count: None,
+ embedded_index: None,
+ file_source: None,
+ value_stats_cols: None,
+ external_path: None,
+ first_row_id: None,
+ write_cols: None,
+ }
+ }
+
+ /// TRUE end-to-end: write a parquet data file containing a
`FixedSizeList<Float32, 2>`
+ /// column, then read it back through `DataFileReader` using a Paimon
`read_type` whose
+ /// field is `DataType::Vector`. This exercises
`build_target_arrow_schema`, the parquet
+ /// format dispatch (by `.parquet` extension), and the read path's
pass-through/cast
+ /// logic — not just a raw Arrow/parquet round-trip.
+ #[tokio::test]
+ async fn test_datafilereader_inline_vector_column_e2e() {
+ // Paimon read schema: a single nullable VECTOR<FLOAT> column of
length 2.
+ let vector_type = VectorType::try_new(true, 2,
DataType::Float(FloatType::new())).unwrap();
+ let read_fields = vec![DataField::new(
+ 0,
+ "embedding".to_string(),
+ DataType::Vector(vector_type),
+ )];
+
+ // Build the physical Arrow data via the Paimon -> Arrow conversion
under test,
+ // so the parquet file matches what the read path expects to
materialize.
+ let arrow_schema = build_target_arrow_schema(&read_fields).unwrap();
+
+ // Build a FixedSizeList<Float32, 2> column:
+ // row 0 = [1.0, 2.0] (non-null)
+ // row 1 = null (null vector row)
+ // row 2 = [3.0, 4.0] (non-null)
+ let mut builder = FixedSizeListBuilder::new(Float32Builder::new(),
2).with_field(Arc::new(
+ ArrowField::new("element", ArrowDataType::Float32, true),
+ ));
+ builder.values().append_value(1.0);
+ builder.values().append_value(2.0);
+ builder.append(true);
+ builder.values().append_value(0.0);
+ builder.values().append_value(0.0);
+ builder.append(false); // null vector row
+ builder.values().append_value(3.0);
+ builder.values().append_value(4.0);
+ builder.append(true);
+ let vec_array = builder.finish();
+ let batch = RecordBatch::try_new(arrow_schema.clone(),
vec![Arc::new(vec_array)]).unwrap();
+
+ // Write the data file as parquet into the split's bucket path.
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table_path = "memory:/vector_inline_e2e";
+ let bucket_path = format!("{table_path}/bucket-0");
+ let file_name = "part-0.parquet";
+ let file_path = format!("{bucket_path}/{file_name}");
+ let output = file_io.new_output(&file_path).unwrap();
+ let mut writer: Box<dyn FormatFileWriter> = Box::new(
+ ParquetFormatWriter::new(&output, arrow_schema.clone(), "zstd", 1)
+ .await
+ .unwrap(),
+ );
+ writer.write(&batch).await.unwrap();
+ let file_size = writer.close().await.unwrap();
+
+ // Build a split whose data file's schema_id matches the table
schema_id, so the
+ // read path uses `read_type` directly (no SchemaManager lookup
needed).
+ let table_schema_id = 1;
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(crate::spec::BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(bucket_path)
+ .with_total_buckets(1)
+ .with_data_files(vec![data_file(
+ file_name,
+ file_size as i64,
+ 3,
+ table_schema_id,
+ )])
+ .build()
+ .unwrap();
+
+ let schema_manager = SchemaManager::new(file_io.clone(),
table_path.to_string());
+ let reader = DataFileReader::new(
+ file_io,
+ schema_manager,
+ table_schema_id,
+ read_fields.clone(),
+ read_fields.clone(),
+ Vec::new(),
+ );
+ let batches = reader
+ .read(&[split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
+ assert_eq!(total_rows, 3);
+ let result = &batches[0];
+ assert_eq!(result.num_columns(), 1);
+ assert_eq!(result.schema().field(0).name(), "embedding");
+
+ // The materialized column must be a FixedSizeListArray with the right
length,
+ // child Float32 values, and null bitmap (one non-null and one null
row).
+ let fsl = result
+ .column(0)
+ .as_any()
+ .downcast_ref::<FixedSizeListArray>()
+ .expect("column should materialize as FixedSizeListArray");
+ assert_eq!(fsl.value_length(), 2);
+ assert!(fsl.is_valid(0));
+ assert!(fsl.is_null(1)); // null vector row preserved through the read
path
+ assert!(fsl.is_valid(2));
+
+ let row0 = fsl.value(0);
+ let floats0 = row0
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .expect("child should be Float32Array");
+ assert_eq!(floats0.values(), &[1.0, 2.0]);
+
+ let row2 = fsl.value(2);
+ let floats2 = row2
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .expect("child should be Float32Array");
+ assert_eq!(floats2.values(), &[3.0, 4.0]);
+ }
+}
