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 2fa52044 Fix primary-key ROW layout for Variant tables (#1171)
2fa52044 is described below
commit 2fa52044c67cb941ad965ebb9db62df184429aba
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 19 18:48:39 2026 +0800
Fix primary-key ROW layout for Variant tables (#1171)
---
crates/paimon/src/spec/schema.rs | 4 +
crates/paimon/src/table/audit_log_table/read.rs | 1 +
crates/paimon/src/table/data_evolution_reader.rs | 2 +
crates/paimon/src/table/data_file_reader.rs | 246 +++++++++++++-
crates/paimon/src/table/format_table_read.rs | 1 +
.../src/table/full_text_search_builder/read.rs | 1 +
.../paimon/src/table/global_index_result/read.rs | 1 +
crates/paimon/src/table/hybrid_search_builder.rs | 2 +
crates/paimon/src/table/kv_file_reader.rs | 177 +++++++++-
crates/paimon/src/table/kv_file_writer.rs | 259 +++++++++++++-
crates/paimon/src/table/mod.rs | 17 +
crates/paimon/src/table/pk_full_text_read.rs | 1 +
.../paimon/src/table/pk_vector_data_file_reader.rs | 1 +
.../src/table/pk_vector_indexed_split_read.rs | 1 +
crates/paimon/src/table/pk_vector_orchestrator.rs | 1 +
crates/paimon/src/table/pk_vector_position_read.rs | 4 +
crates/paimon/src/table/pk_vector_read.rs | 3 +
.../pk_vector_read/residual_row_ranges_tests.rs | 2 +
crates/paimon/src/table/pk_vector_read/tests.rs | 2 +
crates/paimon/src/table/table_read.rs | 2 +
crates/paimon/src/table/table_write.rs | 17 +-
.../fixtures/row_variant_interop/CrossVariant.java | 202 +++++++++++
.../tests/fixtures/row_variant_interop/README.md | 66 ++++
.../tests/fixtures/row_variant_interop/generate.py | 141 ++++++++
.../tests/fixtures/row_variant_interop/java.json | 35 ++
.../tests/fixtures/row_variant_interop/java.row | Bin 0 -> 377 bytes
.../tests/fixtures/row_variant_interop/python.json | 38 +++
.../tests/fixtures/row_variant_interop/python.row | Bin 0 -> 1258 bytes
.../tests/fixtures/row_variant_interop/schema.json | 71 ++++
crates/paimon/tests/row_variant_interop_test.rs | 255 ++++++++++++++
crates/paimon/tests/row_variant_test.rs | 376 +++++++++++++++++++++
docs/src/sql.md | 5 +
32 files changed, 1892 insertions(+), 42 deletions(-)
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 9e54e08d..35e152a1 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -1208,6 +1208,10 @@ fn move_reference(mv: &ColumnMove) ->
crate::Result<&str> {
})
}
+/// Must match Java Paimon's `SpecialFields.KEY_FIELD_ID_START`. Duplicated
+/// primary-key fields belong to reserved ID space, outside all user fields.
+pub(crate) const KEY_FIELD_ID_START: i32 = i32::MAX / 2;
+
pub const ROW_ID_FIELD_NAME: &str = "_ROW_ID";
pub const ROW_ID_FIELD_ID: i32 = i32::MAX - 5;
diff --git a/crates/paimon/src/table/audit_log_table/read.rs
b/crates/paimon/src/table/audit_log_table/read.rs
index c4bf56b1..c5e55fc3 100644
--- a/crates/paimon/src/table/audit_log_table/read.rs
+++ b/crates/paimon/src/table/audit_log_table/read.rs
@@ -109,6 +109,7 @@ impl<'a> AuditLogRead<'a> {
self.read.table.schema_manager().clone(),
self.read.table.schema().id(),
self.read.table.schema.fields().to_vec(),
+ self.read.table.data_fields_extractor(),
read_type.clone(),
self.read.data_predicates.clone(),
)
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index cef4da9f..77ac3f90 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -488,6 +488,7 @@ impl DataEvolutionReader {
self.schema_manager.clone(),
self.table_schema_id,
self.table_fields.clone(),
+ |fields| Ok(fields.to_vec()),
self.wide_file_read_type.clone(),
if push_down_raw_predicates {
self.predicates.clone()
@@ -1781,6 +1782,7 @@ fn open_source_stream(
schema_manager,
table_schema_id,
table_fields,
+ |fields| Ok(fields.to_vec()),
source.read_fields().to_vec(),
Vec::new(),
)
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 5c670590..841e347e 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -126,6 +126,8 @@ impl FileRead for TimedFileRead {
}
}
+type DataFieldsExtractor = dyn Fn(&[DataField]) ->
crate::Result<Vec<DataField>> + Send + Sync;
+
/// Reads data files through their format-specific readers.
#[derive(Clone)]
pub(crate) struct DataFileReader {
@@ -133,6 +135,7 @@ pub(crate) struct DataFileReader {
schema_manager: SchemaManager,
table_schema_id: i64,
table_fields: Vec<DataField>,
+ fields_extractor: Arc<DataFieldsExtractor>,
partition_fields: Vec<DataField>,
read_type: Vec<DataField>,
predicates: Vec<Predicate>,
@@ -154,6 +157,7 @@ impl DataFileReader {
schema_manager: SchemaManager,
table_schema_id: i64,
table_fields: Vec<DataField>,
+ fields_extractor: impl Fn(&[DataField]) ->
crate::Result<Vec<DataField>> + Send + Sync + 'static,
read_type: Vec<DataField>,
predicates: Vec<Predicate>,
) -> Self {
@@ -162,6 +166,7 @@ impl DataFileReader {
schema_manager,
table_schema_id,
table_fields,
+ fields_extractor: Arc::new(fields_extractor),
partition_fields: Vec::new(),
read_type,
predicates,
@@ -188,6 +193,20 @@ impl DataFileReader {
self
}
+ fn physical_data_fields(
+ &self,
+ file_meta: &DataFileMeta,
+ file_fields: &[DataField],
+ supplied_fields: Option<&[DataField]>,
+ ) -> crate::Result<Vec<DataField>> {
+ data_schema_fields_for_file(
+ file_fields,
+ file_meta.write_cols.as_deref(),
+ supplied_fields,
+ self.fields_extractor.as_ref(),
+ )
+ }
+
pub(crate) fn with_blob_parallelism(mut self, blob_parallelism: usize) ->
Self {
debug_assert!(blob_parallelism > 0);
self.blob_parallelism = blob_parallelism;
@@ -545,11 +564,8 @@ impl DataFileReader {
let partition_fields = self.partition_fields.clone();
let predicates = self.predicates.clone();
let file_fields = data_fields.clone().unwrap_or_else(||
table_fields.clone());
- let data_schema_fields = data_schema_fields_for_file(
- &file_fields,
- file_meta.write_cols.as_deref(),
- data_schema_fields.as_deref(),
- )?;
+ let data_schema_fields =
+ self.physical_data_fields(&file_meta, &file_fields,
data_schema_fields.as_deref())?;
let projects_row_id = read_type
.iter()
.any(|field| field.name() == ROW_ID_FIELD_NAME);
@@ -991,8 +1007,7 @@ impl DataFileReader {
let target_schema = crate::arrow::build_read_arrow_schema(&read_type)?;
let file_fields = data_fields.clone().unwrap_or_else(||
table_fields.clone());
- let data_schema_fields =
- data_schema_fields_for_file(&file_fields,
file_meta.write_cols.as_deref(), None)?;
+ let data_schema_fields = self.physical_data_fields(&file_meta,
&file_fields, None)?;
// What the reader is asked for.
let projected_read_fields: Vec<DataField> = if data_fields.is_some()
|| file_meta.write_cols.is_some()
@@ -1385,11 +1400,15 @@ fn data_schema_fields_for_file(
file_fields: &[DataField],
write_cols: Option<&[String]>,
data_schema_fields: Option<&[DataField]>,
+ fields_extractor: &DataFieldsExtractor,
) -> crate::Result<Vec<DataField>> {
if let Some(write_cols) = write_cols {
return super::data_evolution_fields::project_by_paths(file_fields,
write_cols);
}
- Ok(data_schema_fields.unwrap_or(file_fields).to_vec())
+ match data_schema_fields {
+ Some(fields) => Ok(fields.to_vec()),
+ None => fields_extractor(file_fields),
+ }
}
/// Convert ranges from their read-path coordinate system to file-local ranges.
@@ -1728,19 +1747,29 @@ mod row_tests {
field(1, "id", DataType::Int(IntType::new())),
field(2, "value", DataType::Int(IntType::new())),
];
- let physical = vec![field(1_000_001, "_KEY_id",
DataType::Int(IntType::new()))];
+ let physical = vec![field(
+ 1 + crate::spec::KEY_FIELD_ID_START,
+ "_KEY_id",
+ DataType::Int(IntType::new()),
+ )];
+ let extractor = |fields: &[DataField]|
Ok(fields.iter().rev().cloned().collect());
assert_eq!(
- data_schema_fields_for_file(&fields, None, None).unwrap(),
- fields
+ data_schema_fields_for_file(&fields, None, None,
&extractor).unwrap(),
+ fields.iter().rev().cloned().collect::<Vec<_>>()
);
- let selected = data_schema_fields_for_file(&fields, None,
Some(&physical)).unwrap();
+ let selected =
+ data_schema_fields_for_file(&fields, None, Some(&physical),
&extractor).unwrap();
assert_eq!(
selected.iter().map(DataField::name).collect::<Vec<_>>(),
vec!["_KEY_id"]
);
- let partial =
- data_schema_fields_for_file(&fields, Some(&["value".to_string()]),
Some(&physical))
- .unwrap();
+ let partial = data_schema_fields_for_file(
+ &fields,
+ Some(&["value".to_string()]),
+ Some(&physical),
+ &extractor,
+ )
+ .unwrap();
assert_eq!(
partial.iter().map(DataField::name).collect::<Vec<_>>(),
vec!["value"]
@@ -2065,6 +2094,7 @@ mod row_tests {
SchemaManager::new(file_io, table_path.to_string()),
schema_id,
fields,
+ |fields| Ok(fields.to_vec()),
read_type,
Vec::new(),
);
@@ -2140,6 +2170,7 @@ mod row_tests {
SchemaManager::new(file_io, table_path.to_string()),
schema_id,
fields,
+ |fields| Ok(fields.to_vec()),
read_type,
vec![predicate],
);
@@ -2208,6 +2239,7 @@ mod row_tests {
SchemaManager::new(file_io.clone(), table_path.to_string()),
1,
fields.clone(),
+ |fields| Ok(fields.to_vec()),
vec![fields[0].clone()],
vec![predicate],
);
@@ -2224,6 +2256,171 @@ mod row_tests {
}
}
+ #[tokio::test]
+ async fn
row_variant_extraction_after_schema_evolution_uses_local_positions() {
+ use crate::spec::VariantType;
+ use crate::variant::GenericVariant;
+ use arrow_array::{BinaryArray, StructArray};
+ use arrow_buffer::NullBuffer;
+
+ let physical_fields = vec![
+ field(7, "old_id", DataType::Int(IntType::new())),
+ field(41, "old_payload", DataType::Variant(VariantType::new())),
+ ];
+ let schema = build_target_arrow_schema(&physical_fields).unwrap();
+ let variants = [
+ Some(r#"{"x":7,"items":[1,9]}"#),
+ Some(r#"{"x":99}"#),
+ Some("null"),
+ None,
+ Some(r#"{"x":88}"#),
+ Some(r#"{"x":"invalid","items":[]}"#),
+ ]
+ .into_iter()
+ .map(|json| json.map(|json| GenericVariant::parse_json(json).unwrap()))
+ .collect::<Vec<_>>();
+ let arrow_schema::DataType::Struct(children) =
schema.field(1).data_type() else {
+ unreachable!()
+ };
+ let payload = StructArray::new(
+ children.clone(),
+ vec![
+ Arc::new(BinaryArray::from_iter_values(variants.iter().map(
+ |value| value.as_ref().map_or(&[][..],
GenericVariant::value),
+ ))),
+ Arc::new(BinaryArray::from_iter_values(variants.iter().map(
+ |value| value.as_ref().map_or(&[][..],
GenericVariant::metadata),
+ ))),
+ ],
+ Some(NullBuffer::from(
+ variants.iter().map(Option::is_some).collect::<Vec<_>>(),
+ )),
+ );
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5, 6])),
+ Arc::new(payload),
+ ],
+ )
+ .unwrap();
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table_path = "memory:/row_variant_local_evolution";
+ let bucket_path = format!("{table_path}/bucket-0");
+ let file_name = "part.row";
+ let output = file_io
+ .new_output(&format!("{bucket_path}/{file_name}"))
+ .unwrap();
+ let mut writer = create_format_writer(&output, schema, "zstd", 1,
None, None, None)
+ .await
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ let file_size = writer.close().await.unwrap().file_size as i64;
+ let mut file = data_file(file_name, file_size, 6, 1);
+ file.first_row_id = Some(1_000);
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(bucket_path)
+ .with_total_buckets(1)
+ .with_data_files(vec![file.clone()])
+ .build()
+ .unwrap();
+
+ let current_fields = vec![
+ field(7, "id", DataType::BigInt(BigIntType::new())),
+ field(41, "payload", DataType::Variant(VariantType::new())),
+ ];
+ let extraction = variant_extraction_row(
+ true,
+ vec![
+ (
+ DataType::Int(IntType::new()),
+ "$.x".into(),
+ false,
+ "UTC".into(),
+ ),
+ (
+ DataType::Int(IntType::new()),
+ "$.items[1]".into(),
+ false,
+ "UTC".into(),
+ ),
+ (
+ DataType::Int(IntType::new()),
+ "$.missing.deep".into(),
+ false,
+ "UTC".into(),
+ ),
+ ],
+ )
+ .unwrap();
+ let reader = DataFileReader::new(
+ file_io.clone(),
+ SchemaManager::new(file_io, table_path.to_string()),
+ 2,
+ current_fields.clone(),
+ |fields| Ok(fields.to_vec()),
+ vec![
+ field(41, "payload", DataType::Row(extraction)),
+ current_fields[0].clone(),
+ ],
+ vec![],
+ );
+ let batches = reader
+ .read_single_file_stream_local(
+ &split,
+ file,
+ Some(physical_fields),
+ None,
+ vec![0, 2, 3, 5],
+ )
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+ let result = &batches[0];
+ assert_eq!(result.num_rows(), 4);
+ assert_eq!(result.schema().field(0).name(), "payload");
+ assert_eq!(
+ result
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap(),
+ &Int64Array::from(vec![1, 3, 4, 6]),
+ );
+ let extracted = result
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert_eq!(
+ extracted
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap(),
+ &Int32Array::from(vec![Some(7), None, None, None]),
+ );
+ assert_eq!(
+ extracted
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap(),
+ &Int32Array::from(vec![Some(9), None, None, None]),
+ );
+ assert_eq!(extracted.column(2).null_count(), 4);
+ assert!(
+ extracted.is_valid(1),
+ "Variant null retains its parent value"
+ );
+ assert!(extracted.is_null(2), "SQL null has no parent Variant value");
+ }
+
/// The predicate must not renumber a projected `_ROW_ID`: the surviving
row
/// keeps its original physical position even when the predicate column is
/// absent from the requested projection.
@@ -2283,6 +2480,7 @@ mod row_tests {
SchemaManager::new(file_io, table_path),
1,
fields,
+ |fields| Ok(fields.to_vec()),
read_type,
vec![predicate],
);
@@ -2435,6 +2633,7 @@ mod row_tests {
SchemaManager::new(file_io.clone(), table_path.clone()),
2,
current.clone(),
+ |fields| Ok(fields.to_vec()),
projected.clone(),
vec![predicate],
);
@@ -2714,6 +2913,7 @@ mod tests {
schema_manager.clone(),
1,
fields.clone(),
+ |fields| Ok(fields.to_vec()),
fields.clone(),
vec![],
);
@@ -2725,6 +2925,7 @@ mod tests {
schema_manager.clone(),
1,
fields.clone(),
+ |fields| Ok(fields.to_vec()),
fields.clone(),
vec![crate::spec::Predicate::AlwaysTrue],
);
@@ -2741,6 +2942,7 @@ mod tests {
schema_manager.clone(),
1,
fields.clone(),
+ |fields| Ok(fields.to_vec()),
fields.clone(),
vec![filtering],
);
@@ -2754,6 +2956,7 @@ mod tests {
schema_manager.clone(),
1,
fields.clone(),
+ |fields| Ok(fields.to_vec()),
fields.clone(),
vec![],
)
@@ -3026,6 +3229,7 @@ mod tests {
SchemaManager::new(file_io.clone(),
table_path.to_string()),
1,
read_fields.clone(),
+ |fields| Ok(fields.to_vec()),
read_fields.clone(),
predicates.clone(),
)
@@ -3123,6 +3327,7 @@ mod tests {
schema_manager,
table_schema_id,
read_fields.clone(),
+ |fields| Ok(fields.to_vec()),
read_fields.clone(),
Vec::new(),
);
@@ -3294,6 +3499,7 @@ mod tests {
schema_manager.clone(),
0,
fields.clone(),
+ |fields| Ok(fields.to_vec()),
fields.clone(),
vec![predicate.clone()],
)
@@ -3310,6 +3516,7 @@ mod tests {
schema_manager,
0,
fields.clone(),
+ |fields| Ok(fields.to_vec()),
fields,
vec![predicate],
)
@@ -3397,6 +3604,7 @@ mod tests {
schema_manager.clone(),
0,
fields.clone(),
+ |fields| Ok(fields.to_vec()),
fields.clone(),
vec![predicate.clone()],
)
@@ -3485,6 +3693,7 @@ mod tests {
schema_manager,
0,
fields.clone(),
+ |fields| Ok(fields.to_vec()),
fields,
vec![predicate],
)
@@ -3557,6 +3766,7 @@ mod tests {
schema_manager,
current_schema.id(),
current_schema.fields().to_vec(),
+ |fields| Ok(fields.to_vec()),
current_schema.fields().to_vec(),
vec![predicate],
)
@@ -3829,6 +4039,7 @@ mod tests {
schema_manager.clone(),
current_schema.id(),
current_schema.fields().to_vec(),
+ |fields| Ok(fields.to_vec()),
vec![current_schema.fields()[0].clone()],
vec![predicate.clone()],
)
@@ -3884,6 +4095,7 @@ mod tests {
schema_manager,
table_schema_id,
fields.clone(),
+ |fields| Ok(fields.to_vec()),
fields.clone(),
Vec::new(),
);
@@ -3962,6 +4174,7 @@ mod tests {
schema_manager,
table_schema_id,
fields.clone(),
+ |fields| Ok(fields.to_vec()),
fields.clone(),
Vec::new(),
);
@@ -4031,6 +4244,7 @@ mod tests {
schema_manager,
table_schema_id,
fields.clone(),
+ |fields| Ok(fields.to_vec()),
read_type,
vec![predicate],
);
@@ -4178,6 +4392,7 @@ mod vector_parquet_tests {
schema_manager,
table_schema_id,
read_fields.clone(),
+ |fields| Ok(fields.to_vec()),
read_fields.clone(),
Vec::new(),
);
@@ -4301,6 +4516,7 @@ mod vector_parquet_tests {
SchemaManager::new(file_io, table_path.to_string()),
schema_id,
read_fields.clone(),
+ |fields| Ok(fields.to_vec()),
read_fields,
predicates,
);
diff --git a/crates/paimon/src/table/format_table_read.rs
b/crates/paimon/src/table/format_table_read.rs
index e8a0fea4..a691bacb 100644
--- a/crates/paimon/src/table/format_table_read.rs
+++ b/crates/paimon/src/table/format_table_read.rs
@@ -163,6 +163,7 @@ impl<'a> FormatTableRead<'a> {
schema_manager.clone(),
schema_id,
data_table_fields.clone(),
+ |fields| Ok(fields.to_vec()),
data_read_type.clone(),
data_predicates.clone(),
)
diff --git a/crates/paimon/src/table/full_text_search_builder/read.rs
b/crates/paimon/src/table/full_text_search_builder/read.rs
index 6cee3304..d24227cd 100644
--- a/crates/paimon/src/table/full_text_search_builder/read.rs
+++ b/crates/paimon/src/table/full_text_search_builder/read.rs
@@ -105,6 +105,7 @@ impl FullTextRead {
plan.table.schema_manager().clone(),
plan.table.schema().id(),
plan.table.schema().fields().to_vec(),
+ plan.table.data_fields_extractor(),
plan.table.schema().fields().to_vec(),
Vec::new(),
)
diff --git a/crates/paimon/src/table/global_index_result/read.rs
b/crates/paimon/src/table/global_index_result/read.rs
index 5cf02b53..229d886b 100644
--- a/crates/paimon/src/table/global_index_result/read.rs
+++ b/crates/paimon/src/table/global_index_result/read.rs
@@ -86,6 +86,7 @@ impl<'a> GlobalIndexResultReadBuilder<'a> {
table.schema_manager().clone(),
table.schema().id(),
table.schema().fields().to_vec(),
+ table.data_fields_extractor(),
read_type,
Vec::new(),
)
diff --git a/crates/paimon/src/table/hybrid_search_builder.rs
b/crates/paimon/src/table/hybrid_search_builder.rs
index bd6f8b9e..7970e8de 100644
--- a/crates/paimon/src/table/hybrid_search_builder.rs
+++ b/crates/paimon/src/table/hybrid_search_builder.rs
@@ -596,6 +596,7 @@ impl<'a> HybridSearchBuilder<'a> {
self.table.schema_manager().clone(),
self.table.schema().id(),
self.table.schema().fields().to_vec(),
+ self.table.data_fields_extractor(),
self.table.schema().fields().to_vec(),
Vec::new(),
)
@@ -762,6 +763,7 @@ impl<'a> HybridSearchBuilder<'a> {
table.schema_manager().clone(),
table.schema().id(),
table.schema().fields().to_vec(),
+ table.data_fields_extractor(),
table.schema().fields().to_vec(),
Vec::new(),
)
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index 44c9a07a..eda7a27c 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -40,7 +40,7 @@ use crate::io::FileIO;
use crate::spec::{
is_variant_extraction_row, BigIntType, CoreOptions, DataField,
DataFileMeta,
DataType as PaimonDataType, MergeEngine, PartialUpdateConfig, Predicate,
TinyIntType,
- SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_ID,
+ KEY_FIELD_ID_START, SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME,
VALUE_KIND_FIELD_ID,
VALUE_KIND_FIELD_NAME,
};
use crate::table::schema_manager::SchemaManager;
@@ -186,7 +186,42 @@ fn ensure_merge_input_limit(input_stream_count: usize,
limit: Option<usize>) ->
/// Java's `KeyValueFieldsExtractor` builds the physical file layout from the
/// schema that wrote the file, including its historical key names and types.
-fn key_value_data_schema_fields(
+pub(super) fn key_value_fields_extractor(
+ table_fields: &[DataField],
+ primary_keys: &[String],
+) -> impl Fn(&[DataField]) -> crate::Result<Vec<DataField>> + Send + Sync +
'static {
+ let key_ids = primary_keys
+ .iter()
+ .map(|name| {
+ table_fields
+ .iter()
+ .find(|field| field.name() == name)
+ .map(DataField::id)
+ })
+ .collect::<Option<Vec<_>>>();
+ move |file_fields| {
+ let key_ids = key_ids.as_ref().ok_or_else(|| Error::DataInvalid {
+ message: "KV key field is absent from the table
schema".to_string(),
+ source: None,
+ })?;
+ let names = key_ids
+ .iter()
+ .map(|id| {
+ file_fields
+ .iter()
+ .find(|field| field.id() == *id)
+ .map(|field| field.name().to_string())
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!("KV key field id {id} is absent from
the file schema"),
+ source: None,
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ key_value_data_schema_fields(file_fields, &names)
+ }
+}
+
+pub(super) fn key_value_data_schema_fields(
file_fields: &[DataField],
trimmed_primary_keys: &[String],
) -> crate::Result<Vec<DataField>> {
@@ -203,7 +238,7 @@ fn key_value_data_schema_fields(
field
.clone()
.with_name(format!("_KEY_{name}"))
- .with_id(field.id() + 1_000_000),
+ .with_id(field.id() + KEY_FIELD_ID_START),
);
}
physical.push(DataField::new(
@@ -683,6 +718,7 @@ impl KeyValueFileReader {
config.schema_manager.clone(),
table_schema_id,
config.table_fields.clone(),
+ key_value_fields_extractor(&config.table_fields,
&config.primary_keys),
internal_read_type.clone(),
pushdown_predicates.clone(),
)
@@ -1119,7 +1155,7 @@ mod tests {
"value"
]
);
- assert_eq!(physical[0].id(), 1_000_000);
+ assert_eq!(physical[0].id(), i32::MAX / 2);
assert_eq!(physical[0].description(), Some("sort key"));
assert_eq!(&physical[3..], fields);
assert!(key_value_data_schema_fields(&fields,
&["id".to_string()]).is_err());
@@ -1189,27 +1225,134 @@ mod tests {
file.file_size = file_size;
file.min_key = int_key(7);
file.max_key = int_key(7);
- let split = DataSplitBuilder::new()
- .with_snapshot(1)
- .with_partition(BinaryRow::new(0))
- .with_bucket(0)
- .with_bucket_path(bucket_path)
- .with_total_buckets(1)
- .with_data_files(vec![file])
- .with_raw_convertible(false)
+ file.delete_row_count = Some(0);
+ for raw_convertible in [false, true] {
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(bucket_path.clone())
+ .with_total_buckets(1)
+ .with_data_files(vec![file.clone()])
+ .with_raw_convertible(raw_convertible)
+ .build()
+ .unwrap();
+ let batches = table
+ .new_read_builder()
+ .new_read()
+ .unwrap()
+ .to_arrow(std::slice::from_ref(&split))
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(int_column(&batches, "id"), vec![7]);
+ assert_eq!(int_column(&batches, "value"), vec![42]);
+
+ let selected = DataFileReader::new(
+ file_io.clone(),
+ table.schema_manager().clone(),
+ table.schema().id(),
+ table.schema().fields().to_vec(),
+ table.data_fields_extractor(),
+ table.schema().fields().to_vec(),
+ Vec::new(),
+ )
+ .read_single_file_stream_local(
+ &split,
+ file.clone(),
+ Some(old_schema.fields().to_vec()),
+ None,
+ vec![0],
+ )
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(int_column(&selected, "id"), vec![7]);
+ assert_eq!(int_column(&selected, "value"), vec![42]);
+ }
+ }
+
+ #[tokio::test]
+ async fn kv_row_nested_predicate_preserves_high_user_field_ids() {
+ let file_io = test_file_io();
+ let table_path = "memory:/kv_row_high_user_field_ids";
+ setup_dirs(&file_io, table_path).await;
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "profile",
+ DataType::Row(crate::spec::RowType::new(vec![
+ DataField::new(0, "score".into(),
DataType::Int(IntType::new())),
+ DataField::new(1, "payload".into(),
DataType::Variant(VariantType::new())),
+ ])),
+ )
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("file.format", "row")
.build()
.unwrap();
- let batches = table
- .new_read_builder()
+ // A resolved historical schema can have sparse field IDs after many
+ // schema changes. This ID is valid user space, below Java's reserved
+ // key/system range, and must not alias the duplicated key column.
+ let mut json = serde_json::to_value(TableSchema::new(0,
&schema)).unwrap();
+ json["fields"][1]["id"] = serde_json::json!(1_000_000);
+ json["highestFieldId"] = serde_json::json!(1_000_000);
+ let schema: TableSchema = serde_json::from_value(json).unwrap();
+ schema.validate_resolved_structure().unwrap();
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "kv_row_high_user_field_ids"),
+ table_path.to_string(),
+ schema,
+ None,
+ );
+ write_schema_file(&table, table.schema()).await;
+ let variant =
crate::variant::GenericVariant::parse_json(r#"{"x":7}"#).unwrap();
+ let ArrowDataType::Struct(variant_fields) = variant_arrow_type() else {
+ unreachable!()
+ };
+ let payload = StructArray::new(
+ variant_fields,
+ vec![
+ Arc::new(BinaryArray::from(vec![variant.value(); 2])),
+ Arc::new(BinaryArray::from(vec![variant.metadata(); 2])),
+ ],
+ None,
+ );
+ let arrow_schema =
build_target_arrow_schema(table.schema().fields()).unwrap();
+ let ArrowDataType::Struct(profile_fields) =
arrow_schema.field(1).data_type() else {
+ unreachable!()
+ };
+ let profile = StructArray::new(
+ profile_fields.clone(),
+ vec![Arc::new(Int32Array::from(vec![-5, 42])), Arc::new(payload)],
+ None,
+ );
+ let batch = RecordBatch::try_new(
+ arrow_schema,
+ vec![Arc::new(Int32Array::from(vec![7, 8])), Arc::new(profile)],
+ )
+ .unwrap();
+ write_commit(&table, &batch).await;
+
+ let predicates = PredicateBuilder::new(table.schema().fields());
+ let score = predicates.nested_field("profile", &["score"]).unwrap();
+ let mut builder = table.new_read_builder();
+ builder.with_filter(predicates.greater_than(&score,
Datum::Int(0)).unwrap());
+ let plan = builder.new_scan().plan().await.unwrap();
+ assert!(!plan.splits().is_empty());
+ assert!(plan.splits().iter().all(|split| split.raw_convertible()));
+ let batches = builder
.new_read()
.unwrap()
- .to_arrow(&[split])
+ .to_arrow(plan.splits())
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
- assert_eq!(int_column(&batches, "id"), vec![7]);
- assert_eq!(int_column(&batches, "value"), vec![42]);
+ assert_eq!(int_column(&batches, "id"), vec![8]);
}
#[tokio::test]
diff --git a/crates/paimon/src/table/kv_file_writer.rs
b/crates/paimon/src/table/kv_file_writer.rs
index 33845d4e..cdd8b33c 100644
--- a/crates/paimon/src/table/kv_file_writer.rs
+++ b/crates/paimon/src/table/kv_file_writer.rs
@@ -20,9 +20,9 @@
//! Buffers data in memory, sorts by primary key on flush, and prepends
//! `_SEQUENCE_NUMBER` and `_VALUE_KIND` columns.
//!
-//! Uses thin-mode (`data-file.thin-mode`): the physical file schema is
-//! `[_SEQUENCE_NUMBER, _VALUE_KIND, all_user_cols...]` — primary key columns
-//! are NOT duplicated. The read path extracts keys from the value portion.
+//! Columnar files use thin-mode: `[_SEQUENCE_NUMBER, _VALUE_KIND,
all_user_cols...]`.
+//! Positional ROW files prepend duplicated `_KEY_*` columns, matching Java's
+//! key-value file schema. Internal merge, index and statistics batches stay
thin.
//!
//! Reference:
[org.apache.paimon.io.KeyValueDataFileWriterImpl](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriterImpl.java)
@@ -36,7 +36,7 @@ use crate::spec::stats::{compute_column_stats,
BinaryTableStats};
use crate::spec::{
data_file_to_file_index_file_name, extract_datum_from_arrow,
AggregationConfig, BigIntType,
BinaryRowBuilder, CoreOptions, DataField, DataFileMeta, DataType,
MergeEngine,
- PartialUpdateConfig, RowKind, TinyIntType, SEQUENCE_NUMBER_FIELD_ID,
+ PartialUpdateConfig, RowKind, TinyIntType, KEY_FIELD_ID_START,
SEQUENCE_NUMBER_FIELD_ID,
SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_ID, VALUE_KIND_FIELD_NAME,
};
use crate::table::data_file_index_writer::FileIndexOptions;
@@ -126,6 +126,27 @@ struct IndexedFileWrite<'a> {
delete_row_count: i64,
}
+struct RowKeyProjection {
+ schema: Arc<ArrowSchema>,
+ column_indices: Vec<usize>,
+}
+
+impl RowKeyProjection {
+ fn project(&self, batch: &RecordBatch) -> Result<RecordBatch> {
+ let columns = self
+ .column_indices
+ .iter()
+ .map(|&index| Arc::clone(batch.column(index)))
+ .collect();
+ RecordBatch::try_new(Arc::clone(&self.schema),
columns).map_err(|error| {
+ crate::Error::DataInvalid {
+ message: format!("Failed to prepend ROW key columns: {error}"),
+ source: None,
+ }
+ })
+ }
+}
+
impl KeyValueFileWriter {
pub(crate) fn new(
file_io: FileIO,
@@ -194,10 +215,10 @@ impl KeyValueFileWriter {
/// Statistics retain logical value fields independently of the physical
KV schema.
fn new_format_writer_factory(
&self,
- schema: arrow_schema::SchemaRef,
+ mut schema: arrow_schema::SchemaRef,
is_changelog: bool,
) -> Result<Arc<dyn FormatWriterFactory>> {
- let physical_fields = build_physical_fields(&schema,
&self.config.value_fields)?;
+ let mut physical_fields = build_physical_fields(&schema,
&self.config.value_fields)?;
let mut options = self.config.table_options.clone();
let mode = CoreOptions::new(&options)
.pk_file_metadata_stats_mode(0, is_changelog)?
@@ -208,6 +229,27 @@ impl KeyValueFileWriter {
} else {
&self.config.file_format
};
+ if file_format.eq_ignore_ascii_case("row") {
+ let projection = self.row_key_projection(&schema)?;
+ let key_count = self.config.primary_key_indices.len();
+ physical_fields = projection
+ .column_indices
+ .iter()
+ .enumerate()
+ .map(|(index, &source)| {
+ let field = physical_fields[source].clone();
+ if index < key_count {
+ let id = field.id() + KEY_FIELD_ID_START;
+ field
+ .with_id(id)
+
.with_name(projection.schema.field(index).name().clone())
+ } else {
+ field
+ }
+ })
+ .collect();
+ schema = projection.schema;
+ }
create_format_writer_factory(
file_format,
schema,
@@ -219,6 +261,42 @@ impl KeyValueFileWriter {
)
}
+ fn row_key_projection(&self, schema: &Arc<ArrowSchema>) ->
Result<RowKeyProjection> {
+ let mut fields = Vec::new();
+ let mut column_indices = Vec::new();
+ // These indices contain only trimmed keys; partition key columns are
+ // stored in the value portion but do not belong in the key prefix.
+ for &index in &self.config.primary_key_indices {
+ let key = &self.config.value_fields[index];
+ let source =
+ schema
+ .index_of(key.name())
+ .map_err(|error| crate::Error::DataInvalid {
+ message: format!(
+ "ROW key '{}' is absent from the physical schema:
{error}",
+ key.name()
+ ),
+ source: None,
+ })?;
+ let arrow_field = schema.field(source);
+ fields.push(Arc::new(ArrowField::new(
+ format!("_KEY_{}", key.name()),
+ arrow_field.data_type().clone(),
+ arrow_field.is_nullable(),
+ )));
+ column_indices.push(source);
+ }
+ fields.extend(schema.fields().iter().cloned());
+ column_indices.extend(0..schema.fields().len());
+ Ok(RowKeyProjection {
+ schema: Arc::new(ArrowSchema::new_with_metadata(
+ fields,
+ schema.metadata().clone(),
+ )),
+ column_indices,
+ })
+ }
+
pub(crate) fn with_resources(mut self, resources: Option<ResourceContext>)
-> Self {
self.buffer_reservation =
resources.as_ref().map(ResourceContext::reservation);
self.resources = resources;
@@ -516,6 +594,11 @@ impl KeyValueFileWriter {
.transpose()?
};
let physical_schema = build_physical_schema(&user_schema);
+ let row_projection = write
+ .file_format
+ .eq_ignore_ascii_case("row")
+ .then(|| self.row_key_projection(&physical_schema))
+ .transpose()?;
let mut blob_references = ManagedBlobReferences::new(
&self.config.value_fields,
&CoreOptions::new(&self.config.table_options),
@@ -604,7 +687,14 @@ impl KeyValueFileWriter {
let _ = self.file_io.delete_file(&file_path).await;
return Err(error);
}
- if let Err(error) = writer.write(&chunk_batch).await {
+ let row_batch = row_projection
+ .as_ref()
+ .map(|projection| projection.project(&chunk_batch))
+ .transpose()?;
+ if let Err(error) = writer
+ .write(row_batch.as_ref().unwrap_or(&chunk_batch))
+ .await
+ {
let _ = writer.close().await;
let _ = self.file_io.delete_file(&file_path).await;
return Err(error);
@@ -1383,6 +1473,161 @@ mod tests {
}
}
+ #[tokio::test]
+ async fn row_key_prefix_is_applied_independently_to_data_and_changelog() {
+ use crate::arrow::format::create_format_reader;
+ use crate::arrow::{build_target_arrow_schema, variant_arrow_type};
+ use crate::spec::{BinaryRow, VariantType};
+ use crate::variant::GenericVariant;
+ use arrow_array::{BinaryArray, StructArray};
+ use arrow_buffer::NullBuffer;
+ use futures::TryStreamExt;
+
+ for (data_format, changelog_format) in
+ [("row", "row"), ("row", "parquet"), ("parquet", "row")]
+ {
+ let io = FileIOBuilder::new("memory").build().unwrap();
+ let mut config = test_write_config(MergeEngine::Deduplicate);
+ config.file_format = data_format.into();
+ config.changelog_file_format = changelog_format.into();
+ config.input_changelog = true;
+ config.write_buffer_size = i64::MAX;
+ // The full PK includes the partition column, but only the trimmed
+ // `id` key belongs in the physical key prefix.
+ config.primary_keys.push("seq".into());
+ config.value_fields[2] =
+ DataField::new(2, "value".into(),
DataType::Variant(VariantType::new()));
+ let value_fields = config.value_fields.clone();
+ let variants = [Some("[1,2]"), Some("null"), Some(r#"{"x":7}"#),
None]
+ .into_iter()
+ .map(|json| json.map(|json|
GenericVariant::parse_json(json).unwrap()))
+ .collect::<Vec<_>>();
+ let ArrowDataType::Struct(children) = variant_arrow_type() else {
+ unreachable!()
+ };
+ let values = StructArray::new(
+ children,
+ vec![
+ Arc::new(BinaryArray::from_iter_values(variants.iter().map(
+ |value| value.as_ref().map_or(&[][..],
GenericVariant::value),
+ ))),
+ Arc::new(BinaryArray::from_iter_values(variants.iter().map(
+ |value| value.as_ref().map_or(&[][..],
GenericVariant::metadata),
+ ))),
+ ],
+ Some(NullBuffer::from(
+ variants.iter().map(Option::is_some).collect::<Vec<_>>(),
+ )),
+ );
+ let batch = RecordBatch::try_new(
+ build_target_arrow_schema(&value_fields).unwrap(),
+ vec![
+ Arc::new(Int32Array::from(vec![3, 2, 1, 4])),
+ Arc::new(Int64Array::from(vec![7; 4])),
+ Arc::new(values),
+ ],
+ )
+ .unwrap();
+ let mut writer = KeyValueFileWriter::new(io.clone(), config,
0).unwrap();
+ writer.write(&batch).await.unwrap();
+ let prepared = writer.prepare_commit().await.unwrap();
+ for (files, format) in [
+ (&prepared.data_files, data_format),
+ (&prepared.changelog_files, changelog_format),
+ ] {
+ assert_eq!(files.len(), 1);
+ let file = &files[0];
+ let path = file.data_file_path(writer.paths.bucket_path());
+ let mut physical = vec![
+ DataField::new(
+ SEQUENCE_NUMBER_FIELD_ID,
+ SEQUENCE_NUMBER_FIELD_NAME.into(),
+ DataType::BigInt(BigIntType::new()),
+ ),
+ DataField::new(
+ VALUE_KIND_FIELD_ID,
+ VALUE_KIND_FIELD_NAME.into(),
+ DataType::TinyInt(TinyIntType::new()),
+ ),
+ ];
+ physical.extend(value_fields.clone());
+ if format == "row" {
+ physical.insert(
+ 0,
+ value_fields[0]
+ .clone()
+ .with_name("_KEY_id".into())
+ .with_id(KEY_FIELD_ID_START),
+ );
+ let bytes =
io.new_input(&path).unwrap().read().await.unwrap();
+ assert_eq!(bytes[bytes.len() - 8], 1);
+ } else {
+ // Dense Parquet stats omit Variant values and identify the
+ // remaining logical columns, independently of KV metadata.
+ assert_eq!(file.value_stats_cols, Some(vec!["id".into(),
"seq".into()]));
+ assert_eq!(file.value_stats.null_counts(), &vec![Some(0),
Some(0)]);
+ }
+ assert_eq!(file.key_stats.null_counts().len(), 1);
+ let min =
BinaryRow::from_serialized_bytes(&file.min_key).unwrap();
+ let max =
BinaryRow::from_serialized_bytes(&file.max_key).unwrap();
+ assert_eq!(min.get_int(0).unwrap(), 1);
+ assert_eq!(max.get_int(0).unwrap(), 4);
+
+ let input = io.new_input(&path).unwrap();
+ let batches = create_format_reader(&path, false, &physical)
+ .unwrap()
+ .read_batch_stream(
+ Box::new(input.reader().await.unwrap()),
+ file.file_size as u64,
+ &physical,
+ None,
+ None,
+ None,
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+ let stored = &batches[0];
+ assert_eq!(stored.num_columns(), if format == "row" { 6 } else
{ 5 });
+ if format == "row" {
+ assert_eq!(
+ stored.column_by_name("_KEY_id").unwrap().to_data(),
+ stored.column_by_name("id").unwrap().to_data(),
+ );
+ assert!(stored.column_by_name("_KEY_seq").is_none());
+ }
+ let values = stored
+ .column_by_name("value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let value_bytes = values
+ .column(0)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ let metadata = values
+ .column(1)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ for (row, original) in [2, 1, 0, 3].into_iter().enumerate() {
+ if let Some(expected) = &variants[original] {
+ assert!(values.is_valid(row));
+ assert_eq!(value_bytes.value(row), expected.value());
+ assert_eq!(metadata.value(row), expected.metadata());
+ } else {
+ assert!(values.is_null(row));
+ }
+ }
+ }
+ }
+ }
+
#[tokio::test]
async fn variant_noop_plans_preserve_pk_schema_and_changelog() {
use crate::arrow::{build_target_arrow_schema, variant_arrow_type};
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index cbbd883d..f11c3426 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -404,6 +404,23 @@ impl Table {
&self.schema
}
+ pub(crate) fn data_fields_extractor(
+ &self,
+ ) -> impl Fn(&[DataField]) -> Result<Vec<DataField>> + Send + Sync +
'static {
+ let primary_key = !self.schema.primary_keys().is_empty();
+ let key_value_fields = kv_file_reader::key_value_fields_extractor(
+ self.schema.fields(),
+ &self.schema.trimmed_primary_keys(),
+ );
+ move |fields| {
+ if primary_key {
+ key_value_fields(fields)
+ } else {
+ Ok(fields.to_vec())
+ }
+ }
+ }
+
/// Get the FileIO instance for this table.
pub fn file_io(&self) -> &FileIO {
&self.file_io
diff --git a/crates/paimon/src/table/pk_full_text_read.rs
b/crates/paimon/src/table/pk_full_text_read.rs
index e83504db..e808bc30 100644
--- a/crates/paimon/src/table/pk_full_text_read.rs
+++ b/crates/paimon/src/table/pk_full_text_read.rs
@@ -756,6 +756,7 @@ mod read_tests {
schema_manager,
1,
id_fields(),
+ |fields| Ok(fields.to_vec()),
id_fields(),
Vec::new(),
);
diff --git a/crates/paimon/src/table/pk_vector_data_file_reader.rs
b/crates/paimon/src/table/pk_vector_data_file_reader.rs
index 99bd319c..4af8d833 100644
--- a/crates/paimon/src/table/pk_vector_data_file_reader.rs
+++ b/crates/paimon/src/table/pk_vector_data_file_reader.rs
@@ -452,6 +452,7 @@ mod integration_tests {
schema_manager,
table_schema_id,
read_fields.clone(),
+ |fields| Ok(fields.to_vec()),
read_fields.clone(),
Vec::new(),
);
diff --git a/crates/paimon/src/table/pk_vector_indexed_split_read.rs
b/crates/paimon/src/table/pk_vector_indexed_split_read.rs
index 7fefb6f1..74df0f4b 100644
--- a/crates/paimon/src/table/pk_vector_indexed_split_read.rs
+++ b/crates/paimon/src/table/pk_vector_indexed_split_read.rs
@@ -549,6 +549,7 @@ mod e2e_tests {
schema_manager,
1,
id_fields(),
+ |fields| Ok(fields.to_vec()),
id_fields(),
Vec::new(),
);
diff --git a/crates/paimon/src/table/pk_vector_orchestrator.rs
b/crates/paimon/src/table/pk_vector_orchestrator.rs
index 2f750b73..0c6aa60b 100644
--- a/crates/paimon/src/table/pk_vector_orchestrator.rs
+++ b/crates/paimon/src/table/pk_vector_orchestrator.rs
@@ -1041,6 +1041,7 @@ mod e2e_tests {
schema_manager,
1,
id_fields(),
+ |fields| Ok(fields.to_vec()),
id_fields(),
Vec::new(),
)
diff --git a/crates/paimon/src/table/pk_vector_position_read.rs
b/crates/paimon/src/table/pk_vector_position_read.rs
index 06721816..dbc720dc 100644
--- a/crates/paimon/src/table/pk_vector_position_read.rs
+++ b/crates/paimon/src/table/pk_vector_position_read.rs
@@ -495,6 +495,7 @@ mod tests {
schema_manager,
schema_id,
id_fields(),
+ |fields| Ok(fields.to_vec()),
read_type,
predicates,
);
@@ -938,6 +939,7 @@ mod tests {
schema_manager,
1,
id_fields(),
+ |fields| Ok(fields.to_vec()),
id_fields(),
vec![predicate],
);
@@ -984,6 +986,7 @@ mod tests {
schema_manager,
1,
id_fields(),
+ |fields| Ok(fields.to_vec()),
vec![reserved],
Vec::new(),
);
@@ -1031,6 +1034,7 @@ mod tests {
schema_manager,
1,
id_fields(),
+ |fields| Ok(fields.to_vec()),
vec![id_field(), requested_row_id],
Vec::new(),
);
diff --git a/crates/paimon/src/table/pk_vector_read.rs
b/crates/paimon/src/table/pk_vector_read.rs
index 38210bab..37382da5 100644
--- a/crates/paimon/src/table/pk_vector_read.rs
+++ b/crates/paimon/src/table/pk_vector_read.rs
@@ -298,6 +298,7 @@ async fn search_pk_raw_candidates_batch_with_plan(
table.schema_manager().clone(),
table.schema().id(),
table.schema().fields().to_vec(),
+ table.data_fields_extractor(),
vec![vector_field.clone()],
Vec::new(),
)
@@ -548,6 +549,7 @@ async fn search_pk_candidates_batch_with_plan(
table.schema_manager().clone(),
table.schema().id(),
table.schema().fields().to_vec(),
+ table.data_fields_extractor(),
vec![vector_field.clone()],
Vec::new(),
)
@@ -606,6 +608,7 @@ pub(super) async fn filtered_row_ranges(
table.schema_manager().clone(),
table.schema().id(),
table.schema().fields().to_vec(),
+ table.data_fields_extractor(),
residual_read_type,
Vec::new(),
)
diff --git
a/crates/paimon/src/table/pk_vector_read/residual_row_ranges_tests.rs
b/crates/paimon/src/table/pk_vector_read/residual_row_ranges_tests.rs
index f50ef9d0..ba533e07 100644
--- a/crates/paimon/src/table/pk_vector_read/residual_row_ranges_tests.rs
+++ b/crates/paimon/src/table/pk_vector_read/residual_row_ranges_tests.rs
@@ -163,6 +163,7 @@ async fn build_reader_and_split(
SchemaManager::new(file_io, table_path.to_string()),
1,
vec![id_field()],
+ |fields| Ok(fields.to_vec()),
vec![id_field(), row_id_field()],
Vec::new(),
);
@@ -423,6 +424,7 @@ async fn build_reader_and_split_no_first_row_id(
SchemaManager::new(file_io, table_path.to_string()),
1,
vec![id_field()],
+ |fields| Ok(fields.to_vec()),
vec![id_field(), row_id_field()],
Vec::new(),
);
diff --git a/crates/paimon/src/table/pk_vector_read/tests.rs
b/crates/paimon/src/table/pk_vector_read/tests.rs
index a31f0003..10fa3ee3 100644
--- a/crates/paimon/src/table/pk_vector_read/tests.rs
+++ b/crates/paimon/src/table/pk_vector_read/tests.rs
@@ -280,6 +280,7 @@ async fn vector_rerank_fixture(
schema_manager,
schema_id,
read_fields.clone(),
+ |fields| Ok(fields.to_vec()),
read_fields,
Vec::new(),
);
@@ -593,6 +594,7 @@ async fn
rerank_aligns_recomputed_distance_by_position_column() {
schema_manager,
schema_id,
read_fields.clone(),
+ |fields| Ok(fields.to_vec()),
read_fields.clone(),
Vec::new(),
);
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index ffcd4bcf..7fe76656 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -814,6 +814,7 @@ impl<'a> PaimonTableRead<'a> {
self.table.schema_manager().clone(),
self.table.schema().id(),
self.table.schema.fields().to_vec(),
+ self.table.data_fields_extractor(),
scan_type,
predicates,
)
@@ -1880,6 +1881,7 @@ impl<'a> PaimonTableRead<'a> {
self.table.schema_manager().clone(),
self.table.schema().id(),
self.table.schema.fields().to_vec(),
+ self.table.data_fields_extractor(),
self.read_type().to_vec(),
self.data_predicates.clone(),
)
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 67c6eb87..8513c2be 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -3289,7 +3289,7 @@ pub(in crate::table) mod tests {
bucket_dir_name(messages[0].bucket),
file.file_name
);
- let physical_fields = vec![
+ let mut physical_fields = vec![
DataField::new(
SEQUENCE_NUMBER_FIELD_ID,
SEQUENCE_NUMBER_FIELD_NAME.into(),
@@ -3303,6 +3303,11 @@ pub(in crate::table) mod tests {
table.schema().fields()[0].clone(),
table.schema().fields()[1].clone(),
];
+ if format == "row" {
+ let key = table.schema().fields()[0].clone();
+ let key_id = key.id() + crate::spec::KEY_FIELD_ID_START;
+ physical_fields.insert(0,
key.with_name("_KEY_id".into()).with_id(key_id));
+ }
let format_reader = create_format_reader(&file_path, false,
&physical_fields).unwrap();
let input = table.file_io().new_input(&file_path).unwrap();
let stream = format_reader
@@ -3318,8 +3323,14 @@ pub(in crate::table) mod tests {
.unwrap();
let batches: Vec<RecordBatch> =
futures::TryStreamExt::try_collect(stream).await.unwrap();
- assert_eq!(collect_i32(&batches, 2), vec![1, 2], "{format}");
- assert_eq!(collect_i32(&batches, 3), vec![10, 20], "{format}");
+ let id_index = batches[0].schema().index_of("id").unwrap();
+ let value_index = batches[0].schema().index_of("value").unwrap();
+ assert_eq!(collect_i32(&batches, id_index), vec![1, 2],
"{format}");
+ assert_eq!(collect_i32(&batches, value_index), vec![10, 20],
"{format}");
+ if format == "row" {
+ let key_index =
batches[0].schema().index_of("_KEY_id").unwrap();
+ assert_eq!(collect_i32(&batches, key_index), vec![1, 2]);
+ }
TableCommit::new(table.clone(), "test-user".into())
.commit(messages)
.await
diff --git a/crates/paimon/tests/fixtures/row_variant_interop/CrossVariant.java
b/crates/paimon/tests/fixtures/row_variant_interop/CrossVariant.java
new file mode 100644
index 00000000..2850fa6c
--- /dev/null
+++ b/crates/paimon/tests/fixtures/row_variant_interop/CrossVariant.java
@@ -0,0 +1,202 @@
+/*
+ * 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.
+ */
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.StringJoiner;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.DataGetters;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.data.InternalMap;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.variant.GenericVariant;
+import org.apache.paimon.data.variant.Variant;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FormatReaderContext;
+import org.apache.paimon.format.FormatReaderFactory;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.FileRecordReader;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.TinyIntType;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.types.VariantType;
+
+public class CrossVariant {
+ static RowType type() {
+ return new RowType(
+ Arrays.asList(
+ new DataField(SpecialFields.KEY_FIELD_ID_START,
"_KEY_id", new IntType()),
+ new DataField(2147483646, "_SEQUENCE_NUMBER", new
BigIntType()),
+ new DataField(2147483645, "_VALUE_KIND", new
TinyIntType()),
+ new DataField(0, "id", new IntType()),
+ new DataField(1, "payload", new VariantType()),
+ new DataField(
+ 2,
+ "nested",
+ new RowType(
+ Arrays.asList(
+ new DataField(10, "payload",
new VariantType()),
+ new DataField(11, "note", new
VarCharType(100))))),
+ new DataField(3, "items", new ArrayType(new
VariantType())),
+ new DataField(
+ 4, "attrs", new MapType(new VarCharType(100),
new VariantType())),
+ new DataField(5, "extra", new VariantType())));
+ }
+
+ static String[] jsons = {
+ "{\"x\":1,\"n\":{\"items\":[true,null,\"unicode\\u4e2d\"]}}",
+ "[1,{\"x\":\"value\"},null]",
+ "null",
+ null,
+ "\"scalar\"",
+ "{}",
+ "[]",
+ "true",
+ "1234567890123456789"
+ };
+
+ static String hex(byte[] b) {
+ StringBuilder s = new StringBuilder();
+ for (byte v : b) s.append(String.format("%02x", v & 255));
+ return s.toString();
+ }
+
+ static String hex(java.nio.ByteBuffer buffer) {
+ java.nio.ByteBuffer c = buffer.duplicate();
+ byte[] b = new byte[c.remaining()];
+ c.get(b);
+ return hex(b);
+ }
+
+ static String variant(Variant v) {
+ return hex(v.valueBuffer()) + ":" + hex(v.metadataBuffer());
+ }
+
+ static void dump(InternalRow row, RowType t) {
+ StringBuilder s = new StringBuilder();
+ for (int i = 0; i < t.getFieldCount(); i++) {
+ if (i > 0) s.append('|');
+ s.append(field(row, i, t.getTypeAt(i)));
+ }
+ System.out.println(s);
+ }
+
+ static String field(DataGetters d, int i, DataType t) {
+ if (d.isNullAt(i)) return "SQLNULL";
+ switch (t.getTypeRoot()) {
+ case INTEGER:
+ return "I" + d.getInt(i);
+ case BIGINT:
+ return "L" + d.getLong(i);
+ case TINYINT:
+ return "B" + d.getByte(i);
+ case VARCHAR:
+ return "S" + d.getString(i);
+ case VARIANT:
+ return variant(d.getVariant(i));
+ case ROW:
+ {
+ RowType r = (RowType) t;
+ InternalRow a = d.getRow(i, r.getFieldCount());
+ StringJoiner s = new StringJoiner(",", "R[", "]");
+ for (int j = 0; j < r.getFieldCount(); j++) s.add(field(a,
j, r.getTypeAt(j)));
+ return s.toString();
+ }
+ case ARRAY:
+ {
+ InternalArray a = d.getArray(i);
+ DataType e = ((ArrayType) t).getElementType();
+ StringJoiner s = new StringJoiner(",", "A[", "]");
+ for (int j = 0; j < a.size(); j++) s.add(field(a, j, e));
+ return s.toString();
+ }
+ case MAP:
+ {
+ InternalMap a = d.getMap(i);
+ MapType m = (MapType) t;
+ StringJoiner s = new StringJoiner(",", "M[", "]");
+ for (int j = 0; j < a.size(); j++)
+ s.add(
+ field(a.keyArray(), j, m.getKeyType())
+ + "="
+ + field(a.valueArray(), j,
m.getValueType()));
+ return s.toString();
+ }
+ default:
+ throw new RuntimeException(t.toString());
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ RowType t = type();
+ Path path = new Path(args[1]);
+ LocalFileIO io = new LocalFileIO();
+ FileFormat f = FileFormat.fromIdentifier("row", new Options());
+ if (args[0].equals("write")) {
+ PositionOutputStream out = io.newOutputStream(path, false);
+ FormatWriter w = f.createWriterFactory(t).create(out, "zstd");
+ for (int i = 0; i < jsons.length; i++) {
+ Variant v = jsons[i] == null ? null :
GenericVariant.fromJson(jsons[i]);
+ Variant n = GenericVariant.fromJson("null");
+ Map<BinaryString, Variant> m = new LinkedHashMap<>();
+ m.put(BinaryString.fromString("a"), v);
+ m.put(BinaryString.fromString("b"), null);
+ m.put(BinaryString.fromString("c"), n);
+ InternalRow row =
+ GenericRow.of(
+ i,
+ (long) i,
+ (byte) 0,
+ i,
+ v,
+ i == 3 ? null : GenericRow.of(v,
BinaryString.fromString("ok")),
+ i == 5 ? null : new GenericArray(new Object[]
{v, null, n}),
+ i == 6 ? null : new GenericMap(m),
+ v);
+ w.addElement(row);
+ dump(row, t);
+ }
+ w.close();
+ out.close();
+ } else {
+ FormatReaderFactory factory = f.createReaderFactory(t, t, new
ArrayList<>());
+ FileRecordReader<InternalRow> reader =
+ factory.createReader(
+ new FormatReaderContext(io, path,
io.getFileSize(path), null, null));
+ reader.forEachRemaining(row -> dump(row, t));
+ reader.close();
+ }
+ }
+}
diff --git a/crates/paimon/tests/fixtures/row_variant_interop/README.md
b/crates/paimon/tests/fixtures/row_variant_interop/README.md
new file mode 100644
index 00000000..1ba98b2f
--- /dev/null
+++ b/crates/paimon/tests/fixtures/row_variant_interop/README.md
@@ -0,0 +1,66 @@
+# ROW v1 Variant interoperability fixtures
+
+These files were written with Apache Paimon Java and Python from PR #10650
+(`e4aa5dc9a5cc3c9e8bb7ff48df7b521e036d9eda`), then read and rewritten by Rust.
+Both actual Java and Python readers checked the Rust output. No Rust encoder
+was used to construct the input fixtures.
+
+`java.row` is written by `CrossVariant.java` using `FileFormat`'s ROW writer.
+`python.row` is written by `generate.py` using Python's `FormatRowWriter`.
+The two fixtures deliberately use different Variant object builders and values.
+Their logical table schema is `schema.json`; their physical PK layout is:
+
+```text
+_KEY_id, _SEQUENCE_NUMBER, _VALUE_KIND, id, payload, nested, items, attrs,
extra
+```
+
+The key ID comes from Java's `SpecialFields.KEY_FIELD_ID_START` (1073741823),
+with sequence/kind IDs 2147483646/2147483645. ROW v1 stores fields positionally
+and does not embed field IDs. Nine columns exercise a null bitmap crossing a
+byte boundary. The block uses ZSTD and the footer retains version 1.
+
+The nine rows cover objects, arrays, JSON null, SQL null (Java), strings, empty
+objects/arrays, booleans, a large integer, Unicode, and a Python object with
+130 keys and long strings. Every Variant is also repeated under ROW, ARRAY,
+and MAP, with SQL-null containers/elements and JSON-null elements. The small
+`java.json`/`python.json` files record SHA256 of each root Variant's original
+value and metadata buffers. Repeated children must have the same buffers;
+this avoids storing large repeated hex dumps. SQL null is represented by a
+null digest record, whereas JSON null has two concrete buffer digests.
+
+The Rust test reads the canonical PK layout, checks all buffer digests and
+container validity, writes through Rust's native PK writer, commits, and
+reads the resulting table again. Run it without a Java/Python dependency:
+
+```sh
+cargo test -p paimon --test row_variant_interop_test --locked
+```
+
+To regenerate or cross-check with an adjacent Paimon checkout, first compile
+and run `RowFormatReadWriteTest` there so its Surefire report supplies the
+resolved Java classpath. Python needs the checkout's normal PyArrow/ZSTD
+requirements, and `javac`/`java` must be installed:
+
+```sh
+mvn -pl paimon-format -am -DwildcardSuites=none \
+ -Dtest=RowFormatReadWriteTest -DfailIfNoTests=false test
+```
+
+From the Rust checkout:
+
+```sh
+python3 crates/paimon/tests/fixtures/row_variant_interop/generate.py \
+ --paimon-repo /path/to/paimon --generate
+
+PAIMON_ROW_INTEROP_OUTPUT=/tmp/paimon-row-interop \
+ cargo test -p paimon --test row_variant_interop_test --locked
+python3 crates/paimon/tests/fixtures/row_variant_interop/generate.py \
+ --paimon-repo /path/to/paimon --check-rust-output /tmp/paimon-row-interop
+```
+
+The external check reads Rust's exported files independently with Java and
+Python and compares keys, row kinds, Variant buffers, and null containers.
+Rust's generated sequence numbers are checked by those readers but are not
+required to equal the sequence numbers in the input fixtures. Regenerating
+with another ZSTD version can change compressed bytes without changing the
+ROW payload or the recorded Variant digests.
diff --git a/crates/paimon/tests/fixtures/row_variant_interop/generate.py
b/crates/paimon/tests/fixtures/row_variant_interop/generate.py
new file mode 100644
index 00000000..a01e8ff8
--- /dev/null
+++ b/crates/paimon/tests/fixtures/row_variant_interop/generate.py
@@ -0,0 +1,141 @@
+# 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.
+
+"""Regenerate fixtures or check Rust output with actual Java/Python readers."""
+import argparse
+import hashlib
+import json
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import xml.etree.ElementTree as ET
+
+parser = argparse.ArgumentParser(description=__doc__)
+parser.add_argument('--paimon-repo', required=True, type=Path)
+parser.add_argument('--generate', action='store_true')
+parser.add_argument('--check-rust-output', type=Path)
+args = parser.parse_args()
+sys.path.insert(0, str(args.paimon_repo / 'paimon-python'))
+
+import pyarrow as pa
+from pypaimon.data.generic_variant import GenericVariant
+from pypaimon.read.reader.format_row_reader import FormatRowReader
+from pypaimon.schema.data_types import AtomicType, DataField, RowType,
PyarrowFieldParser
+from pypaimon.write.writer.format_row_writer import FormatRowWriter
+
+root = Path(__file__).parent
+schema = json.loads((root / 'schema.json').read_text())
+# Java's real key ID is used here; physical IDs are not embedded in ROW files.
+fields = [DataField(1073741823, '_KEY_id', AtomicType('INT', False)),
+ DataField(2147483646, '_SEQUENCE_NUMBER', AtomicType('BIGINT',
False)),
+ DataField(2147483645, '_VALUE_KIND', AtomicType('TINYINT', False))]
+fields += RowType.from_dict({'type': 'ROW', 'fields': schema['fields']}).fields
+
+class FileIO:
+ def get_file_size(self, path):
+ return Path(path).stat().st_size
+
+ def new_input_stream(self, path):
+ return open(path, 'rb')
+
+def read(path):
+ reader = FormatRowReader(FileIO(), str(path), [f.name for f in fields],
fields, None)
+ batches = []
+ while (batch := reader.read_arrow_batch()) is not None:
+ batches.append(batch)
+ reader.close()
+ return pa.Table.from_batches(batches)
+
+def write(path, data):
+ with path.open('wb') as stream:
+ writer = FormatRowWriter(stream, fields)
+ writer.write_table(data)
+ writer.close()
+
+def encoded(value):
+ if value is None:
+ return 'SQLNULL'
+ return value['value'].hex() + ':' + value['metadata'].hex()
+
+def rows(data):
+ result = []
+ for row in data.to_pylist():
+ nested = row['nested']
+ items = row['items']
+ attrs = row['attrs']
+ result.append('|'.join([
+ 'I' + str(row['_KEY_id']), 'L' + str(row['_SEQUENCE_NUMBER']),
+ 'B' + str(row['_VALUE_KIND']), 'I' + str(row['id']),
encoded(row['payload']),
+ 'SQLNULL' if nested is None else 'R[' + encoded(nested['payload'])
+ ',S' + nested['note'] + ']',
+ 'SQLNULL' if items is None else 'A[' + ','.join(encoded(v) for v
in items) + ']',
+ 'SQLNULL' if attrs is None else 'M[' + ','.join('S' + k + '=' +
encoded(v) for k, v in attrs) + ']',
+ encoded(row['extra'])]))
+ return result
+
+def digests(data):
+ return [None if v is None else {
+ 'value': hashlib.sha256(v['value']).hexdigest(),
+ 'metadata': hashlib.sha256(v['metadata']).hexdigest()
+ } for v in data.column('payload').to_pylist()]
+
+report = args.paimon_repo /
'paimon-format/target/surefire-reports/TEST-org.apache.paimon.format.row.RowFormatReadWriteTest.xml'
+properties = ET.parse(report).getroot().find('properties')
+classpath = next(p.attrib['value'] for p in properties if p.attrib.get('name')
== 'java.class.path')
+with tempfile.TemporaryDirectory() as temp:
+ subprocess.run(['javac', '-source', '8', '-target', '8', '-cp', classpath,
+ '-d', temp, str(root / 'CrossVariant.java')], check=True)
+
+ def java(action, path):
+ completed = subprocess.run(['java', '-cp', temp + ':' + classpath,
+ 'CrossVariant', action, str(path)],
+ check=True, capture_output=True, text=True)
+ return completed.stdout.splitlines()
+
+ if args.generate:
+ (root / 'java.row').unlink(missing_ok=True)
+ expected = java('write', root / 'java.row')
+ assert rows(read(root / 'java.row')) == expected
+ values = [{'z': 1, 'a': {'c': [True, None, '中']}}, [1, {'x': 'value'},
None],
+ None, 'scalar', {}, [], True, 1234567890123456789,
+ {f'key_{i:03}': 'x' * 100 for i in range(130)}]
+ variants = [GenericVariant.from_python(value) for value in values]
+ variants = [{'value': v.value(), 'metadata': v.metadata()} for v in
variants]
+ data = []
+ for i, variant in enumerate(variants):
+ data.append({'_KEY_id': i, '_SEQUENCE_NUMBER': i, '_VALUE_KIND':
0, 'id': i,
+ 'payload': variant, 'extra': variant,
+ 'nested': None if i == 3 else {'payload': variant,
'note': 'ok'},
+ 'items': None if i == 5 else [variant, None,
variants[2]],
+ 'attrs': None if i == 6 else [('a', variant), ('b',
None), ('c', variants[2])]})
+ data = pa.Table.from_pylist(data,
schema=PyarrowFieldParser.from_paimon_schema(fields))
+ write(root / 'python.row', data)
+ assert java('read', root / 'python.row') == rows(data)
+ for name in ['java', 'python']:
+ (root / f'{name}.json').write_text(json.dumps(digests(read(root /
f'{name}.row')), indent=2) + '\n')
+ print('Regenerated real Java/Python fixtures and verified both
readers.')
+
+ if args.check_rust_output:
+ for name in ['java', 'python']:
+ actual = rows(read(args.check_rust_output / f'rust_{name}.row'))
+ assert actual == java('read', args.check_rust_output /
f'rust_{name}.row')
+ expected = rows(read(root / f'{name}.row'))
+ # Rust generates its own sequence numbers. All keys, kinds and
+ # logical values must still match byte for byte.
+ without_sequence = lambda records: [r.split('|')[:1] +
r.split('|')[2:] for r in records]
+ assert without_sequence(actual) == without_sequence(expected)
+ print('Rust output matches Java and Python: keys, kinds, all Variant
buffers and nulls.')
diff --git a/crates/paimon/tests/fixtures/row_variant_interop/java.json
b/crates/paimon/tests/fixtures/row_variant_interop/java.json
new file mode 100644
index 00000000..9e33568c
--- /dev/null
+++ b/crates/paimon/tests/fixtures/row_variant_interop/java.json
@@ -0,0 +1,35 @@
+[
+ {
+ "value":
"9c8017286a82e7a16be843dbb352e0fe206533c1dc825686db4921d3e96912bf",
+ "metadata":
"4bc3e4d938c4e5c412b16bce9d59ea7d975aae7467881a75ad7c00a807bc9514"
+ },
+ {
+ "value":
"7e33010e02ad87dd7650f2d43b93484bc6d9113a2f6153f49755d8f4e1eaead4",
+ "metadata":
"ac68724267938add046294d31056841c157a23b33cc6479e3e1c2a1c884d1940"
+ },
+ {
+ "value":
"6e340b9cffb37a989ca544e6bb780a2c78901d3fb33738768511a30617afa01d",
+ "metadata":
"fb50dc0717ff266cf9baf82b1ce7a1c2ef6d9247859680b11a19fb7077f5f222"
+ },
+ null,
+ {
+ "value":
"2287b4bf263e2a414e5a4e2d3f93285157f8a43badf0047c13641018f97aa0c1",
+ "metadata":
"fb50dc0717ff266cf9baf82b1ce7a1c2ef6d9247859680b11a19fb7077f5f222"
+ },
+ {
+ "value":
"070b13c004e9475fa155cd84d90de9f87db1b7dea6707329638a6ce66043d246",
+ "metadata":
"fb50dc0717ff266cf9baf82b1ce7a1c2ef6d9247859680b11a19fb7077f5f222"
+ },
+ {
+ "value":
"3e4479e6b5a0a3bed1440325ca4bb09327a1dfb39c28bdd59de53a687ade0105",
+ "metadata":
"fb50dc0717ff266cf9baf82b1ce7a1c2ef6d9247859680b11a19fb7077f5f222"
+ },
+ {
+ "value":
"e52d9c508c502347344d8c07ad91cbd6068afc75ff6292f062a09ca381c89e71",
+ "metadata":
"fb50dc0717ff266cf9baf82b1ce7a1c2ef6d9247859680b11a19fb7077f5f222"
+ },
+ {
+ "value":
"f65fbda4602f3d25f0a98f85eaa6288180867ea733d7b4cd8e82dee21af88b9a",
+ "metadata":
"fb50dc0717ff266cf9baf82b1ce7a1c2ef6d9247859680b11a19fb7077f5f222"
+ }
+]
diff --git a/crates/paimon/tests/fixtures/row_variant_interop/java.row
b/crates/paimon/tests/fixtures/row_variant_interop/java.row
new file mode 100644
index 00000000..60df8601
Binary files /dev/null and
b/crates/paimon/tests/fixtures/row_variant_interop/java.row differ
diff --git a/crates/paimon/tests/fixtures/row_variant_interop/python.json
b/crates/paimon/tests/fixtures/row_variant_interop/python.json
new file mode 100644
index 00000000..53d5ead1
--- /dev/null
+++ b/crates/paimon/tests/fixtures/row_variant_interop/python.json
@@ -0,0 +1,38 @@
+[
+ {
+ "value":
"97c00231b425a7b6b3176acbe7c8eb28dab6ff7e66a79f6d623e76ecebde9612",
+ "metadata":
"58d4206d06b8f56c52a2010760af11c4b08d1d4bba111953b6d4a6f548c5f2c3"
+ },
+ {
+ "value":
"7e33010e02ad87dd7650f2d43b93484bc6d9113a2f6153f49755d8f4e1eaead4",
+ "metadata":
"ac68724267938add046294d31056841c157a23b33cc6479e3e1c2a1c884d1940"
+ },
+ {
+ "value":
"6e340b9cffb37a989ca544e6bb780a2c78901d3fb33738768511a30617afa01d",
+ "metadata":
"fb50dc0717ff266cf9baf82b1ce7a1c2ef6d9247859680b11a19fb7077f5f222"
+ },
+ {
+ "value":
"2287b4bf263e2a414e5a4e2d3f93285157f8a43badf0047c13641018f97aa0c1",
+ "metadata":
"fb50dc0717ff266cf9baf82b1ce7a1c2ef6d9247859680b11a19fb7077f5f222"
+ },
+ {
+ "value":
"070b13c004e9475fa155cd84d90de9f87db1b7dea6707329638a6ce66043d246",
+ "metadata":
"fb50dc0717ff266cf9baf82b1ce7a1c2ef6d9247859680b11a19fb7077f5f222"
+ },
+ {
+ "value":
"3e4479e6b5a0a3bed1440325ca4bb09327a1dfb39c28bdd59de53a687ade0105",
+ "metadata":
"fb50dc0717ff266cf9baf82b1ce7a1c2ef6d9247859680b11a19fb7077f5f222"
+ },
+ {
+ "value":
"e52d9c508c502347344d8c07ad91cbd6068afc75ff6292f062a09ca381c89e71",
+ "metadata":
"fb50dc0717ff266cf9baf82b1ce7a1c2ef6d9247859680b11a19fb7077f5f222"
+ },
+ {
+ "value":
"f65fbda4602f3d25f0a98f85eaa6288180867ea733d7b4cd8e82dee21af88b9a",
+ "metadata":
"fb50dc0717ff266cf9baf82b1ce7a1c2ef6d9247859680b11a19fb7077f5f222"
+ },
+ {
+ "value":
"5590dd91c553c98e8b526f54e762128731e41066fd38eaeadeea0f859f7b31dc",
+ "metadata":
"16bfef20a04bca47f0abd8904a653039a673ed9946e51fe6b04ac5ac19106bf5"
+ }
+]
diff --git a/crates/paimon/tests/fixtures/row_variant_interop/python.row
b/crates/paimon/tests/fixtures/row_variant_interop/python.row
new file mode 100644
index 00000000..8e5fed89
Binary files /dev/null and
b/crates/paimon/tests/fixtures/row_variant_interop/python.row differ
diff --git a/crates/paimon/tests/fixtures/row_variant_interop/schema.json
b/crates/paimon/tests/fixtures/row_variant_interop/schema.json
new file mode 100644
index 00000000..815f973f
--- /dev/null
+++ b/crates/paimon/tests/fixtures/row_variant_interop/schema.json
@@ -0,0 +1,71 @@
+{
+ "version": 3,
+ "id": 0,
+ "fields": [
+ {
+ "id": 0,
+ "name": "id",
+ "type": "INT NOT NULL"
+ },
+ {
+ "id": 1,
+ "name": "payload",
+ "type": "VARIANT"
+ },
+ {
+ "id": 2,
+ "name": "nested",
+ "type": {
+ "type": "ROW",
+ "fields": [
+ {
+ "id": 10,
+ "name": "payload",
+ "type": "VARIANT"
+ },
+ {
+ "id": 11,
+ "name": "note",
+ "type": "STRING"
+ }
+ ],
+ "nullable": true
+ }
+ },
+ {
+ "id": 3,
+ "name": "items",
+ "type": {
+ "type": "ARRAY",
+ "element": "VARIANT",
+ "nullable": true
+ }
+ },
+ {
+ "id": 4,
+ "name": "attrs",
+ "type": {
+ "type": "MAP",
+ "key": "STRING",
+ "value": "VARIANT",
+ "nullable": true
+ }
+ },
+ {
+ "id": 5,
+ "name": "extra",
+ "type": "VARIANT"
+ }
+ ],
+ "highestFieldId": 11,
+ "partitionKeys": [],
+ "primaryKeys": [
+ "id"
+ ],
+ "options": {
+ "file.format": "row",
+ "bucket": "1"
+ },
+ "comment": null,
+ "timeMillis": 1
+}
diff --git a/crates/paimon/tests/row_variant_interop_test.rs
b/crates/paimon/tests/row_variant_interop_test.rs
new file mode 100644
index 00000000..d00b92a4
--- /dev/null
+++ b/crates/paimon/tests/row_variant_interop_test.rs
@@ -0,0 +1,255 @@
+// 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.
+
+#![cfg(all(feature = "storage-fs", not(windows)))]
+
+use arrow_array::{
+ Array, BinaryArray, Int32Array, ListArray, MapArray, RecordBatch,
StringArray, StructArray,
+};
+use arrow_select::concat::concat_batches;
+use futures::TryStreamExt;
+use paimon::catalog::Identifier;
+use paimon::io::FileIOBuilder;
+use paimon::spec::{BinaryRow, BinaryTableStats, DataFileMeta, TableSchema};
+use paimon::table::DataSplitBuilder;
+use paimon::Table;
+use serde::Deserialize;
+use sha2::{Digest, Sha256};
+
+#[derive(Deserialize)]
+struct VariantDigest {
+ value: String,
+ metadata: String,
+}
+
+fn assert_variant(array: &StructArray, row: usize, expected:
Option<&VariantDigest>) {
+ let Some(expected) = expected else {
+ assert!(array.is_null(row), "SQL null at row {row}");
+ return;
+ };
+ assert!(array.is_valid(row), "non-null Variant at row {row}");
+ for (index, expected) in [&expected.value,
&expected.metadata].iter().enumerate() {
+ let child = array
+ .column(index)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ assert!(child.is_valid(row));
+ assert_eq!(
+ &format!("{:x}", Sha256::digest(child.value(row))),
+ *expected
+ );
+ }
+}
+
+fn assert_values(batch: &RecordBatch, expected: &[Option<VariantDigest>]) {
+ assert_eq!(batch.num_rows(), expected.len());
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let root = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let nested = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let nested_values = nested
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let notes = nested
+ .column(1)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let items = batch
+ .column(3)
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .unwrap();
+ let attrs = batch.column(4).as_any().downcast_ref::<MapArray>().unwrap();
+ let extra = batch
+ .column(5)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let keys = attrs.keys().as_any().downcast_ref::<StringArray>().unwrap();
+ let values = attrs
+ .values()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let variant_null = expected[2].as_ref();
+ assert!(variant_null.is_some());
+ for (row, expected) in expected.iter().enumerate() {
+ assert_eq!(ids.value(row), row as i32);
+ assert_variant(root, row, expected.as_ref());
+ assert_variant(extra, row, expected.as_ref());
+ assert_eq!(nested.is_null(row), row == 3);
+ if nested.is_valid(row) {
+ assert_variant(nested_values, row, expected.as_ref());
+ assert_eq!(notes.value(row), "ok");
+ }
+ assert_eq!(items.is_null(row), row == 5);
+ if items.is_valid(row) {
+ let elements = items.value(row);
+ let elements =
elements.as_any().downcast_ref::<StructArray>().unwrap();
+ assert_eq!(elements.len(), 3);
+ assert_variant(elements, 0, expected.as_ref());
+ assert_variant(elements, 1, None);
+ assert_variant(elements, 2, variant_null);
+ }
+ assert_eq!(attrs.is_null(row), row == 6);
+ if attrs.is_valid(row) {
+ let start = attrs.value_offsets()[row] as usize;
+ assert_eq!(attrs.value_offsets()[row + 1] as usize - start, 3);
+ for (index, key) in ["a", "b", "c"].iter().enumerate() {
+ assert_eq!(keys.value(start + index), *key);
+ }
+ assert_variant(values, start, expected.as_ref());
+ assert_variant(values, start + 1, None);
+ assert_variant(values, start + 2, variant_null);
+ }
+ }
+}
+
+#[tokio::test]
+async fn java_and_python_row_variant_files_round_trip_through_rust() {
+ let fixtures: [(&str, &[u8], &str); 2] = [
+ (
+ "java",
+ include_bytes!("fixtures/row_variant_interop/java.row"),
+ include_str!("fixtures/row_variant_interop/java.json"),
+ ),
+ (
+ "python",
+ include_bytes!("fixtures/row_variant_interop/python.row"),
+ include_str!("fixtures/row_variant_interop/python.json"),
+ ),
+ ];
+ for (name, bytes, expected) in fixtures {
+ let expected: Vec<Option<VariantDigest>> =
serde_json::from_str(expected).unwrap();
+ let schema: TableSchema =
+
serde_json::from_str(include_str!("fixtures/row_variant_interop/schema.json")).unwrap();
+ let temp = tempfile::tempdir().unwrap();
+ std::fs::create_dir(temp.path().join("schema")).unwrap();
+ std::fs::write(
+ temp.path().join("schema/schema-0"),
+ serde_json::to_vec(&schema).unwrap(),
+ )
+ .unwrap();
+ std::fs::write(temp.path().join("fixture.row"), bytes).unwrap();
+ assert_eq!(bytes[bytes.len() - 8], 1, "ROW v1 fixture");
+ let location = format!("file://{}", temp.path().display());
+ let table = Table::new(
+ FileIOBuilder::new("file").build().unwrap(),
+ Identifier::new("default", name),
+ location.clone(),
+ schema,
+ None,
+ );
+ let file = DataFileMeta {
+ file_name: "fixture.row".into(),
+ file_size: bytes.len() as i64,
+ row_count: expected.len() as i64,
+ min_key: Vec::new(),
+ max_key: Vec::new(),
+ key_stats: BinaryTableStats::empty(),
+ value_stats: BinaryTableStats::empty(),
+ min_sequence_number: 0,
+ max_sequence_number: 8,
+ schema_id: 0,
+ level: 0,
+ extra_files: Vec::new(),
+ creation_time: None,
+ delete_row_count: Some(0),
+ embedded_index: None,
+ file_source: None,
+ value_stats_cols: None,
+ external_path: None,
+ first_row_id: None,
+ write_cols: None,
+ column_max_sequence_numbers: None,
+ };
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(location.clone())
+ .with_total_buckets(1)
+ .with_data_files(vec![file])
+ .with_raw_convertible(true)
+ .build()
+ .unwrap();
+ let batches = table
+ .new_read_builder()
+ .new_read()
+ .unwrap()
+ .to_arrow(&[split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let batch = concat_batches(&batches[0].schema(), &batches).unwrap();
+ assert_values(&batch, &expected);
+
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ // Optional output supports manual Java/Python mutual reads without
making
+ // the regular test depend on another language or a writable fixture
path.
+ if let Some(output) = std::env::var_os("PAIMON_ROW_INTEROP_OUTPUT") {
+ let output = std::path::Path::new(&output);
+ std::fs::create_dir_all(output).unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].new_files.len(), 1);
+ let file = &messages[0].new_files[0];
+ let bytes = table
+ .file_io()
+
.new_input(&file.data_file_path(&format!("{location}/bucket-0")))
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ std::fs::write(output.join(format!("rust_{name}.row")),
bytes).unwrap();
+ }
+ writer.close().await;
+ builder.new_commit().commit(messages).await.unwrap();
+ let read_builder = table.new_read_builder();
+ let plan = read_builder.new_scan().plan().await.unwrap();
+ let rewritten = read_builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_values(
+ &concat_batches(&rewritten[0].schema(), &rewritten).unwrap(),
+ &expected,
+ );
+ }
+}
diff --git a/crates/paimon/tests/row_variant_test.rs
b/crates/paimon/tests/row_variant_test.rs
new file mode 100644
index 00000000..e8d6a1ef
--- /dev/null
+++ b/crates/paimon/tests/row_variant_test.rs
@@ -0,0 +1,376 @@
+// 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.
+
+#![cfg(all(feature = "storage-fs", not(windows)))]
+
+use arrow_array::{
+ Array, ArrayRef, BinaryArray, Int32Array, RecordBatch, StringArray,
StructArray,
+};
+use arrow_buffer::NullBuffer;
+use arrow_schema::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
+use arrow_select::concat::concat_batches;
+use futures::TryStreamExt;
+use paimon::arrow::variant_arrow_type;
+use paimon::catalog::{Catalog, Identifier};
+use paimon::common::Options;
+use paimon::spec::{
+ variant_extraction_row, DataField, DataType, IntType, RowType, Schema,
VarCharType, VariantType,
+};
+use paimon::variant::GenericVariant;
+use paimon::{CatalogOptions, FileSystemCatalog, Table};
+use std::collections::HashMap;
+use std::sync::Arc;
+
+async fn create_table(kind: &str) -> (tempfile::TempDir, Table) {
+ let temp = tempfile::tempdir().unwrap();
+ let mut options = Options::new();
+ options.set(
+ CatalogOptions::WAREHOUSE,
+ format!("file://{}", temp.path().display()),
+ );
+ let catalog = FileSystemCatalog::new(options).unwrap();
+ catalog
+ .create_database("default", true, HashMap::new())
+ .await
+ .unwrap();
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", DataType::Variant(VariantType::new()))
+ .column(
+ "nested",
+ DataType::Row(RowType::new(vec![
+ DataField::new(0, "payload".into(),
DataType::Variant(VariantType::new())),
+ DataField::new(
+ 1,
+ "tag".into(),
+ DataType::VarChar(VarCharType::string_type()),
+ ),
+ ])),
+ )
+ .option("file.format", "row");
+ if kind == "primary_key" {
+ schema = schema.primary_key(["id"]).option("bucket", "1");
+ } else if kind == "data_evolution" {
+ schema = schema
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true");
+ }
+ let identifier = Identifier::new("default", kind);
+ catalog
+ .create_table(&identifier, schema.build().unwrap(), false)
+ .await
+ .unwrap();
+ let table = catalog.get_table(&identifier).await.unwrap();
+ (temp, table)
+}
+
+fn variants(values: &[Option<GenericVariant>]) -> StructArray {
+ let ArrowDataType::Struct(fields) = variant_arrow_type() else {
+ unreachable!()
+ };
+ StructArray::new(
+ fields,
+ vec![
+ Arc::new(BinaryArray::from_iter_values(values.iter().map(|value| {
+ value.as_ref().map_or(&[][..], GenericVariant::value)
+ }))),
+ Arc::new(BinaryArray::from_iter_values(values.iter().map(|value| {
+ value.as_ref().map_or(&[][..], GenericVariant::metadata)
+ }))),
+ ],
+ Some(NullBuffer::from(
+ values.iter().map(Option::is_some).collect::<Vec<_>>(),
+ )),
+ )
+}
+
+fn input_batch() -> (RecordBatch, Vec<Option<GenericVariant>>) {
+ let values = [
+ Some(r#"{"x":1,"nested":{"items":[true,null,"text"]}}"#),
+ Some(r#"[1,{"x":"value"},null]"#),
+ Some("null"),
+ None,
+ Some(r#""scalar""#),
+ ]
+ .into_iter()
+ .map(|json| json.map(|json| GenericVariant::parse_json(json).unwrap()))
+ .collect::<Vec<_>>();
+ let nested_values = vec![
+ values[0].clone(),
+ None,
+ values[2].clone(),
+ values[1].clone(),
+ values[4].clone(),
+ ];
+ let nested = StructArray::new(
+ vec![
+ Arc::new(Field::new("payload", variant_arrow_type(), true)),
+ Arc::new(Field::new("tag", ArrowDataType::Utf8, true)),
+ ]
+ .into(),
+ vec![
+ Arc::new(variants(&nested_values)),
+ Arc::new(StringArray::from(vec![
+ "object",
+ "sql-null",
+ "variant-null",
+ "parent-null",
+ "scalar",
+ ])),
+ ],
+ Some(NullBuffer::from(vec![true, true, true, false, true])),
+ );
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ Field::new("id", ArrowDataType::Int32, false),
+ Field::new("payload", variant_arrow_type(), true),
+ Field::new("nested", nested.data_type().clone(), true),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(vec![0, 1, 2, 3, 4])),
+ Arc::new(variants(&values)),
+ Arc::new(nested),
+ ],
+ )
+ .unwrap();
+ (batch, values)
+}
+
+async fn write_and_commit(table: &Table, batch: &RecordBatch) {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(batch).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert!(!messages.is_empty());
+ for file in messages.iter().flat_map(|message| &message.new_files) {
+ assert!(file.file_name.ends_with(".row"));
+ let path = file.data_file_path(&format!("{}/bucket-0",
table.location()));
+ let bytes = table
+ .file_io()
+ .new_input(&path)
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ assert_eq!(bytes[bytes.len() - 8], 1, "ordinary Variant keeps ROW v1");
+ }
+ writer.close().await;
+ builder.new_commit().commit(messages).await.unwrap();
+}
+
+async fn read(table: &Table, projection: &[&str]) -> RecordBatch {
+ let mut builder = table.new_read_builder();
+ builder.with_projection(projection).unwrap();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap_or_else(|err| panic!("reading {} with {projection:?}: {err}",
table.location()));
+ concat_batches(&batches[0].schema(), &batches).unwrap()
+}
+
+fn assert_variants(column: &ArrayRef, expected: &[Option<GenericVariant>]) {
+ let column = column.as_any().downcast_ref::<StructArray>().unwrap();
+ let values = column
+ .column(0)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ let metadata = column
+ .column(1)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ assert_eq!(column.len(), expected.len());
+ for (row, expected) in expected.iter().enumerate() {
+ if let Some(expected) = expected {
+ assert!(column.is_valid(row));
+ let actual = GenericVariant::from_parts(
+ values.value(row).to_vec(),
+ metadata.value(row).to_vec(),
+ )
+ .unwrap();
+ assert_eq!(&actual, expected, "row {row}");
+ assert_eq!(actual.to_json().unwrap(), expected.to_json().unwrap());
+ } else {
+ assert!(column.is_null(row), "row {row}");
+ }
+ }
+}
+
+#[tokio::test]
+async fn
ordinary_variant_row_tables_preserve_values_and_reordered_projection() {
+ for kind in ["append", "primary_key", "data_evolution"] {
+ let (_temp, table) = create_table(kind).await;
+ let (input, values) = input_batch();
+ write_and_commit(&table, &input).await;
+
+ let payload_only = read(&table, &["id", "payload"]).await;
+ assert_variants(payload_only.column(1), &values);
+ let result = read(&table, &["id", "payload", "nested"]).await;
+ assert_eq!(result.num_rows(), 5);
+ assert_eq!(result.column(0).to_data(), input.column(0).to_data());
+ assert_variants(result.column(1), &values);
+ assert_eq!(result.column(2).to_data(), input.column(2).to_data());
+
+ let projected = read(&table, &["payload", "id"]).await;
+ assert_eq!(projected.schema().field(0).name(), "payload");
+ assert_eq!(projected.schema().field(1).name(), "id");
+ assert_variants(projected.column(0), &values);
+ assert_eq!(projected.column(1).to_data(), input.column(0).to_data());
+
+ if kind == "primary_key" {
+ // Overlapping key ranges must use the KV merge reader rather than
+ // the single-file raw shortcut exercised above.
+ write_and_commit(&table, &input).await;
+ let plan =
table.new_read_builder().new_scan().plan().await.unwrap();
+ assert!(!plan.splits().is_empty());
+ assert!(plan.splits().iter().all(|split|
!split.raw_convertible()));
+
+ let merged = read(&table, &["id", "payload", "nested"]).await;
+ assert_eq!(merged.num_rows(), 5);
+ assert_eq!(merged.column(0).to_data(), input.column(0).to_data());
+ assert_variants(merged.column(1), &values);
+ assert_eq!(merged.column(2).to_data(), input.column(2).to_data());
+
+ let projected = read(&table, &["payload", "id"]).await;
+ assert_eq!(projected.schema().field(0).name(), "payload");
+ assert_eq!(projected.schema().field(1).name(), "id");
+ assert_variants(projected.column(0), &values);
+ assert_eq!(projected.column(1).to_data(),
input.column(0).to_data());
+ }
+ }
+}
+
+fn extraction_type() -> DataType {
+ DataType::Row(
+ variant_extraction_row(
+ true,
+ ["$.x", "$.missing"].map(|path| {
+ (
+ DataType::Int(IntType::new()),
+ path.into(),
+ false,
+ "UTC".into(),
+ )
+ }),
+ )
+ .unwrap(),
+ )
+}
+
+async fn assert_variant_extraction_read_type(kind: &str, force_merge: bool) {
+ let (_temp, table) = create_table(kind).await;
+ let (input, _) = input_batch();
+ write_and_commit(&table, &input).await;
+ if force_merge {
+ write_and_commit(&table, &input).await;
+ }
+ let fields = table.schema().fields();
+ let payload = &fields[1];
+ let nested = &fields[2];
+ let DataType::Row(nested_row) = nested.data_type() else {
+ unreachable!()
+ };
+ let nested_payload = &nested_row.fields()[0];
+ let mut builder = table.new_read_builder();
+ builder.with_read_type(vec![
+ DataField::new(
+ nested.id(),
+ nested.name().into(),
+ DataType::Row(RowType::new(vec![DataField::new(
+ nested_payload.id(),
+ nested_payload.name().into(),
+ extraction_type(),
+ )])),
+ ),
+ fields[0].clone(),
+ DataField::new(payload.id(), payload.name().into(), extraction_type()),
+ ]);
+ let plan = builder.new_scan().plan().await.unwrap();
+ if force_merge {
+ assert!(!plan.splits().is_empty());
+ assert!(plan.splits().iter().all(|split| !split.raw_convertible()));
+ }
+ let batches = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let result = concat_batches(&batches[0].schema(), &batches).unwrap();
+ assert_eq!(result.schema().field(0).name(), "nested");
+ assert_eq!(result.schema().field(1).name(), "id");
+ let payload = result
+ .column(2)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let x = payload
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(x, &Int32Array::from(vec![Some(1), None, None, None, None]));
+ assert_eq!(payload.column(1).null_count(), 5);
+ assert!(
+ payload.is_valid(2),
+ "Variant null retains its extraction row"
+ );
+ assert!(payload.is_null(3), "SQL NULL remains a null extraction row");
+
+ let nested = result
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert_eq!(nested.num_columns(), 1);
+ assert!(nested.is_null(3));
+ let nested_payload = nested
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert!(nested_payload.is_null(1));
+ assert!(nested_payload.is_valid(2));
+ assert_eq!(
+ nested_payload
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap(),
+ &Int32Array::from(vec![Some(1), None, None, None, None])
+ );
+ assert_eq!(nested_payload.column(1).null_count(), 5);
+}
+
+#[tokio::test]
+async fn ordinary_variant_row_read_type_extracts_top_level_and_nested_fields()
{
+ for (kind, force_merge) in [
+ ("append", false),
+ ("primary_key", false),
+ ("primary_key", true),
+ ] {
+ assert_variant_extraction_read_type(kind, force_merge).await;
+ }
+}
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 01c6f62f..4eff1e96 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -386,6 +386,11 @@ bytes.
`VARIANT` stores semi-structured data using the same logical value + metadata
binary shape as Paimon Java. Use it for JSON-like fields whose schema may
differ row by row.
+Parquet and ROW files support ordinary `VARIANT` reads and writes. Set
+`'file.format' = 'row'` to use ROW storage. ROW preserves the value and
metadata
+buffers in the Java-compatible v1 layout; extracting a path still reads the
+containing compressed row block. The shredding options below apply to Parquet.
+
Create `VARIANT` columns like ordinary table columns:
```sql
