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 066f8078 [core] Support Java Data Evolution ROW sidecars (#1047)
066f8078 is described below
commit 066f8078e35b6c624168670a075f3f3be599baa8
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 6 21:11:49 2026 +0800
[core] Support Java Data Evolution ROW sidecars (#1047)
---
crates/paimon/src/arrow/format/row.rs | 219 +++++-
crates/paimon/src/spec/core_options.rs | 53 ++
crates/paimon/src/table/data_evolution_reader.rs | 54 +-
crates/paimon/src/table/data_evolution_writer.rs | 9 +
crates/paimon/src/table/data_file_reader.rs | 99 ++-
crates/paimon/src/table/data_file_writer.rs | 226 +++++-
.../src/table/dedicated_format_file_writer.rs | 7 +
crates/paimon/src/table/mod.rs | 1 +
crates/paimon/src/table/row_sidecar.rs | 254 +++++++
.../tests/data_evolution_row_sidecar_test.rs | 815 +++++++++++++++++++++
10 files changed, 1675 insertions(+), 62 deletions(-)
diff --git a/crates/paimon/src/arrow/format/row.rs
b/crates/paimon/src/arrow/format/row.rs
index 22e37aed..ff53b619 100644
--- a/crates/paimon/src/arrow/format/row.rs
+++ b/crates/paimon/src/arrow/format/row.rs
@@ -36,11 +36,11 @@ use arrow_array::builder::{
TimestampMillisecondBuilder, TimestampNanosecondBuilder,
TimestampSecondBuilder,
};
use arrow_array::{
- Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array,
Float32Array,
- Float64Array, Int16Array, Int32Array, Int64Array, Int8Array,
LargeBinaryArray, ListArray,
- MapArray, RecordBatch, RecordBatchOptions, StringArray, StructArray,
Time32MillisecondArray,
- TimestampMicrosecondArray, TimestampMillisecondArray,
TimestampNanosecondArray,
- TimestampSecondArray,
+ Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array,
FixedSizeListArray,
+ Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array,
LargeBinaryArray,
+ ListArray, MapArray, RecordBatch, RecordBatchOptions, StringArray,
StructArray,
+ Time32MillisecondArray, TimestampMicrosecondArray,
TimestampMillisecondArray,
+ TimestampNanosecondArray, TimestampSecondArray,
};
use arrow_buffer::{BooleanBuffer, NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow_schema::{DataType as ArrowDataType, Field, Fields, SchemaRef,
TimeUnit};
@@ -169,6 +169,10 @@ impl FormatFileWriter for RowFormatWriter {
self.block_writer.estimated_size()
}
+ fn pending_rows(&self) -> Option<usize> {
+ Some(self.block_writer.row_count())
+ }
+
async fn flush(&mut self) -> crate::Result<()> {
self.flush_block().await
}
@@ -441,6 +445,10 @@ fn validate_arrow_type_for_row_field(
validate_arrow_type_for_row_field(field_name, child.data_type(),
a.element_type())?;
true
}
+ (ArrowDataType::FixedSizeList(child, size), DataType::Vector(v)) => {
+ validate_arrow_type_for_row_field(field_name, child.data_type(),
v.element_type())?;
+ *size == v.length() as i32
+ }
(ArrowDataType::Map(entries, _), DataType::Map(m)) => {
validate_arrow_map_entries(
field_name,
@@ -565,9 +573,7 @@ fn validate_supported_type(data_type: &DataType) ->
crate::Result<()> {
}
Ok(())
}
- DataType::Vector(_) => Err(Error::Unsupported {
- message: "VectorType is not supported in the .row
format".to_string(),
- }),
+ DataType::Vector(v) => validate_supported_type(v.element_type()),
}
}
@@ -738,10 +744,20 @@ fn write_field_value(
let row = downcast::<StructArray>(array, data_type)?;
write_struct_row(out, row, row_idx, r.fields())?;
}
- DataType::Vector(_) => {
- return Err(Error::Unsupported {
- message: "VectorType is not supported in .row field
serialization".to_string(),
- });
+ DataType::Vector(v) => {
+ let vector = downcast::<FixedSizeListArray>(array, data_type)?;
+ let values = vector.value(row_idx);
+ if values.null_count() != 0 {
+ return Err(Error::DataInvalid {
+ message: ".row VECTOR elements must not be null".into(),
+ source: None,
+ });
+ }
+ // Unlike ARRAY, Java writes no element null bitmap for VECTOR.
+ write_var_u32(out, v.length());
+ for index in 0..values.len() {
+ write_field_value(out, &values, index, v.element_type())?;
+ }
}
}
Ok(())
@@ -1207,6 +1223,12 @@ enum ColumnBuilder {
validities: Vec<bool>,
values: Box<ColumnBuilder>,
},
+ Vector {
+ field: Arc<Field>,
+ size: i32,
+ validities: Vec<bool>,
+ values: Box<ColumnBuilder>,
+ },
Map {
entries_field: Arc<Field>,
offsets: Vec<i32>,
@@ -1315,11 +1337,16 @@ impl ColumnBuilder {
validities: Vec::with_capacity(capacity),
len: 0,
},
- DataType::Vector(_) => {
- return Err(Error::Unsupported {
- message: "VectorType is not supported in .row
ColumnBuilder".to_string(),
- });
- }
+ DataType::Vector(v) => Self::Vector {
+ field: Arc::new(Field::new(
+ "element",
+ paimon_type_to_arrow(v.element_type())?,
+ v.element_type().is_nullable(),
+ )),
+ size: v.length() as i32,
+ validities: Vec::with_capacity(capacity),
+ values: Box::new(ColumnBuilder::new(v.element_type(),
capacity)?),
+ },
})
}
@@ -1342,6 +1369,17 @@ impl ColumnBuilder {
Self::TimestampUs(b) => b.append_null(),
Self::TimestampNs(b) => b.append_null(),
Self::Decimal(b) => b.append_null(),
+ Self::Vector {
+ size,
+ validities,
+ values,
+ ..
+ } => {
+ for _ in 0..*size {
+ values.append_null();
+ }
+ validities.push(false);
+ }
Self::Array {
offsets,
validities,
@@ -1430,6 +1468,29 @@ impl ColumnBuilder {
push_nested_offset(offsets, size)?;
validities.push(true);
}
+ (
+ Self::Vector {
+ size,
+ validities,
+ values,
+ ..
+ },
+ DataType::Vector(v),
+ ) => {
+ let stored_size = input.read_var_u32()?;
+ if stored_size != *size as u32 {
+ return Err(Error::DataInvalid {
+ message: format!(
+ ".row VECTOR length {stored_size} differs from
declared length {size}"
+ ),
+ source: None,
+ });
+ }
+ for _ in 0..stored_size {
+ values.read_append(input, v.element_type())?;
+ }
+ validities.push(true);
+ }
(
Self::Map {
offsets,
@@ -1549,6 +1610,23 @@ impl ColumnBuilder {
source: Some(Box::new(e)),
})?,
),
+ Self::Vector {
+ field,
+ size,
+ validities,
+ values,
+ } => Arc::new(
+ FixedSizeListArray::try_new(
+ field,
+ size,
+ values.finish()?,
+ Some(null_buffer(validities)),
+ )
+ .map_err(|error| Error::DataInvalid {
+ message: format!("Failed to build .row VECTOR: {error}"),
+ source: Some(Box::new(error)),
+ })?,
+ ),
Self::Map {
entries_field,
offsets,
@@ -3396,6 +3474,113 @@ mod tests {
assert_eq!(names.value(2), "nine");
}
+ #[test]
+ fn blob_data_block_matches_java_materialized_descriptor() {
+ use crate::spec::BlobType;
+ // Java RowBlockWriter wrote Blob.fromFile(source, offset=6, length=7)
+ // where source contained "prefixpayloadsuffix".
+ let java_block = decode_hex("00077061796c6f61640000000001000000");
+ let fields = vec![DataField::new(
+ 0,
+ "payload".into(),
+ DataType::Blob(BlobType::new()),
+ )];
+ let schema = build_target_arrow_schema(&fields).unwrap();
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![Arc::new(LargeBinaryArray::from_iter([Some(
+ b"payload".as_slice(),
+ )]))],
+ )
+ .unwrap();
+ let mut writer = RowBlockWriter::new();
+ writer.write_row(&batch, 0, &fields).unwrap();
+ assert_eq!(writer.finish(), java_block);
+ let decoded = decode_row_block(&java_block, &fields, schema,
&[0]).unwrap();
+ assert_eq!(
+ decoded
+ .column(0)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap()
+ .value(0),
+ b"payload"
+ );
+ }
+
+ #[test]
+ fn vector_blocks_match_java_and_decode_parent_nulls() {
+ use crate::spec::{FloatType, VectorType};
+ // Generated by Java RowBlockWriter with VECTOR<2,FLOAT NOT NULL>:
+ // GenericRow.of([1.25f, -2.5f]), GenericRow.of(null).
+ let java_block =
decode_hex("00020000a03f000020c001000000000a00000002000000");
+ let fields = vec![DataField::new(
+ 0,
+ "v".into(),
+ DataType::Vector(
+ VectorType::new(2,
DataType::Float(FloatType::with_nullable(false))).unwrap(),
+ ),
+ )];
+ let schema = build_target_arrow_schema(&fields).unwrap();
+ let mut vectors =
+
arrow_array::builder::FixedSizeListBuilder::new(Float32Builder::new(),
2).with_field(
+ Arc::new(Field::new("element", ArrowDataType::Float32, false)),
+ );
+ vectors.values().append_slice(&[1.25, -2.5]);
+ vectors.append(true);
+ vectors.values().append_slice(&[0.0, 0.0]);
+ vectors.append(false);
+ let batch = RecordBatch::try_new(schema.clone(),
vec![Arc::new(vectors.finish())]).unwrap();
+ let mut writer = RowBlockWriter::new();
+ writer.write_row(&batch, 0, &fields).unwrap();
+ writer.write_row(&batch, 1, &fields).unwrap();
+ assert_eq!(writer.finish(), java_block);
+ let decoded = decode_row_block(&java_block, &fields, schema, &[0,
1]).unwrap();
+ let actual = decoded
+ .column(0)
+ .as_any()
+ .downcast_ref::<FixedSizeListArray>()
+ .unwrap();
+ assert!(actual.is_null(1));
+ let first = actual.value(0);
+ assert_eq!(
+ first
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .unwrap()
+ .values()
+ .as_ref(),
+ &[1.25, -2.5]
+ );
+ // A malformed vector length cannot shift the following fields.
+ let mut malformed = java_block.clone();
+ malformed[1] = 3;
+ assert!(decode_row_block(&malformed, &fields, decoded.schema(),
&[0]).is_err());
+ }
+
+ #[test]
+ fn vector_writes_reject_null_elements_and_wrong_dimensions() {
+ use crate::spec::{FloatType, VectorType};
+ let data_type =
+ DataType::Vector(VectorType::new(2,
DataType::Float(FloatType::new())).unwrap());
+ let mut vectors =
arrow_array::builder::FixedSizeListBuilder::new(Float32Builder::new(), 2);
+ vectors.values().append_value(1.0);
+ vectors.values().append_null();
+ vectors.append(true);
+ let array = Arc::new(vectors.finish()) as ArrayRef;
+ let error = write_field_value(&mut Vec::new(), &array, 0,
&data_type).unwrap_err();
+ assert!(error.to_string().contains("elements must not be null"));
+ assert!(validate_arrow_type_for_row_field(
+ "v",
+ &ArrowDataType::FixedSizeList(
+ Arc::new(Field::new("element", ArrowDataType::Float32, true)),
+ 3,
+ ),
+ &data_type
+ )
+ .is_err());
+ }
+
#[tokio::test]
async fn row_reader_reads_java_generated_fixture() {
let fields = java_fixture_fields();
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index 0d33a760..f67a5e37 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -773,6 +773,59 @@ impl<'a> CoreOptions<'a> {
.unwrap_or(false)
}
+ /// Whether a missing optional read target may fall back to the primary
file.
+ pub fn scan_ignore_lost_file(&self) -> bool {
+ self.options
+ .get("scan.ignore-lost-files")
+ .is_some_and(|value| value.eq_ignore_ascii_case("true"))
+ }
+
+ /// Whether normal Data Evolution files also write an aligned ROW sidecar.
+ pub fn data_evolution_row_sidecar_enabled(&self) -> crate::Result<bool> {
+ match self.options.get("data-evolution.row-sidecar.enabled") {
+ None => Ok(false),
+ Some(value) if value.eq_ignore_ascii_case("true") => Ok(true),
+ Some(value) if value.eq_ignore_ascii_case("false") => Ok(false),
+ Some(value) => Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "data-evolution.row-sidecar.enabled must be true or false,
got {value}"
+ ),
+ }),
+ }
+ }
+
+ /// Largest sparse selection served by a ROW sidecar. Java defaults to
4096.
+ pub fn data_evolution_row_sidecar_max_selected_rows(&self) ->
crate::Result<i64> {
+ let option = "data-evolution.row-sidecar.max-selected-rows";
+ let rows = self.parse_i64_option(option)?.unwrap_or(4096);
+ if rows <= 0 {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!("{option} must be positive, got {rows}"),
+ });
+ }
+ Ok(rows)
+ }
+
+ /// Largest fraction of a file served by a ROW sidecar. Java defaults to
0.05.
+ pub fn data_evolution_row_sidecar_max_selection_ratio(&self) ->
crate::Result<f64> {
+ let option = "data-evolution.row-sidecar.max-selection-ratio";
+ let ratio = match self.options.get(option) {
+ Some(raw) => raw
+ .trim()
+ .parse::<f64>()
+ .map_err(|_| crate::Error::ConfigInvalid {
+ message: format!("Invalid {option}: {raw}"),
+ })?,
+ None => 0.05,
+ };
+ if !(ratio > 0.0 && ratio <= 1.0) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!("{option} must be in (0, 1], got {ratio}"),
+ });
+ }
+ Ok(ratio)
+ }
+
pub fn data_evolution_write_cols_optimization_enabled(&self) -> bool {
self.options
.get("data-evolution.write-cols-optimization.enabled")
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index d30d4020..123a138f 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -451,7 +451,8 @@ impl DataEvolutionReader {
// LIMIT reads resolve views only after the surviving batch has
// been selected. Eagerly scanning every split here can fail on a
// reference beyond the quota and load unrelated upstream blobs.
- let mut blob_view_lookup = if self.limit.is_some() &&
resolve_blob_views {
+ let mut blob_view_lookup = if resolve_blob_views &&
(self.limit.is_some() || splits.iter().any(|split|
+ split.data_files().iter().any(|file|
file.extra_files.iter().any(|name| name.ends_with(".row"))))) {
Some(BlobViewLookup::default())
} else {
self.preload_blob_view_lookup(&splits,
filter_before_blob_resolution)
@@ -533,7 +534,8 @@ impl DataEvolutionReader {
.await?;
let has_row_id = file_meta.first_row_id.is_some();
- let mut effective_row_ranges = if has_row_id {
row_ranges.clone() } else { None };
+ let requested_row_ranges = if has_row_id {
row_ranges.clone() } else { None };
+ let mut effective_row_ranges =
requested_row_ranges.clone();
if read_raw_file_index {
// Independent files can apply a physical row
selection
// before decoding. Column-merge groups
cannot: an older
@@ -615,12 +617,13 @@ impl DataEvolutionReader {
.data_evolution_nested_field_enabled()
|| file_schema_nested_enabled,
)
- .read_single_file_stream(
+ .read_single_file_stream_with_requested_ranges(
&split,
file_meta,
data_fields,
deletion_vector,
effective_row_ranges,
+ requested_row_ranges,
)?;
while remaining != Some(0) {
let Some(batch) = stream.next().await else {
break };
@@ -703,6 +706,7 @@ impl DataEvolutionReader {
&split,
&prepared_group,
effective_row_ranges,
+ row_ranges.clone(),
expected_output_rows,
anchor_deletion_vector,
)?;
@@ -772,7 +776,13 @@ impl DataEvolutionReader {
/// trailing.
fn project_output(&self, filtered: RecordBatch) ->
crate::Result<RecordBatch> {
let final_width = self.output_schema.fields().len();
- if filtered.num_columns() == final_width {
+ if filtered.num_columns() == final_width
+ && !filtered
+ .schema()
+ .fields()
+ .iter()
+ .any(|field| super::row_sidecar::is_blob_data(field))
+ {
return Ok(filtered);
}
let columns = filtered.columns()[..final_width].to_vec();
@@ -811,14 +821,12 @@ impl DataEvolutionReader {
return self.project_output(batch);
}
- if self.limit.is_some() {
- if let (Some(lookup), Some(rest_env)) =
- (blob_view_lookup.as_mut(), self.blob_view_rest_env.clone())
- {
- lookup
- .load_missing(rest_env, &batch, &self.blob_view_fields)
- .await?;
- }
+ if let (Some(lookup), Some(rest_env)) =
+ (blob_view_lookup.as_mut(), self.blob_view_rest_env.clone())
+ {
+ lookup
+ .load_missing(rest_env, &batch, &self.blob_view_fields)
+ .await?;
}
batch = self.resolve_blob_view_columns(batch,
blob_view_lookup.as_ref())?;
let mut batch = if !self.blob_as_descriptor &&
!descriptor_fields.is_empty() {
@@ -950,6 +958,7 @@ impl DataEvolutionReader {
split: &DataSplit,
prepared_group: &PreparedMergeGroup,
row_ranges: Option<Vec<RowRange>>,
+ requested_row_ranges: Option<Vec<RowRange>>,
expected_output_rows: usize,
anchor_deletion_vector: Option<DeletionVectorContext>,
) -> crate::Result<ArrowRecordBatchStream> {
@@ -1065,6 +1074,7 @@ impl DataEvolutionReader {
&split,
source,
row_ranges.clone(),
+ requested_row_ranges.clone(),
file_io.clone(),
schema_manager.clone(),
table_schema_id,
@@ -1192,8 +1202,15 @@ impl DataEvolutionReader {
}
emitted_rows += rows_to_emit;
+ let blob_data_fields =
source_plan.column_plan.iter().enumerate().filter_map(|(idx, provider)| {
+ let (source_idx, field_offset) = provider.as_ref()?;
+ let (batch, _) = source_cursors[*source_idx].as_ref()?;
+
super::row_sidecar::is_blob_data(batch.schema().field(*field_offset))
+ .then(|| target_schema.field(idx).name().clone())
+ }).collect();
+ let merged_schema =
super::row_sidecar::mark_blob_data(&target_schema, &blob_data_fields);
let merged =
- RecordBatch::try_new(target_schema.clone(),
columns).map_err(|e| {
+ RecordBatch::try_new(merged_schema, columns).map_err(|e| {
Error::UnexpectedError {
message: format!("Failed to build merged
RecordBatch: {e}"),
source: Some(Box::new(e)),
@@ -1271,7 +1288,8 @@ where
let mut descriptor_columns = Vec::new();
for (idx, field) in schema.fields().iter().enumerate() {
- if blob_descriptor_fields.contains(field.name()) {
+ if blob_descriptor_fields.contains(field.name()) &&
!super::row_sidecar::is_blob_data(field)
+ {
if let Some(bin_col) = batch
.column(idx)
.as_any()
@@ -1347,7 +1365,7 @@ fn collect_blob_view_structs(
view_structs: &mut HashSet<BlobViewStruct>,
) -> crate::Result<()> {
for (idx, field) in batch.schema().fields().iter().enumerate() {
- if !blob_view_fields.contains(field.name()) {
+ if !blob_view_fields.contains(field.name()) ||
super::row_sidecar::is_blob_data(field) {
continue;
}
let col = large_binary_column(batch, idx, field.name())?;
@@ -1381,7 +1399,7 @@ fn replace_blob_view_columns(
let mut changed = false;
for (idx, field) in schema.fields().iter().enumerate() {
- if !blob_view_fields.contains(field.name()) {
+ if !blob_view_fields.contains(field.name()) ||
super::row_sidecar::is_blob_data(field) {
columns.push(batch.column(idx).clone());
continue;
}
@@ -1627,6 +1645,7 @@ fn open_source_stream(
split: &DataSplit,
source: &FieldSource,
row_ranges: Option<Vec<RowRange>>,
+ requested_row_ranges: Option<Vec<RowRange>>,
file_io: FileIO,
schema_manager: SchemaManager,
table_schema_id: i64,
@@ -1717,12 +1736,13 @@ fn open_source_stream(
file, data_fields, ..
} => {
let deletion_vector = shifted_deletion_vector_for_file(file,
anchor_deletion_vector)?;
- file_reader.read_single_file_stream(
+ file_reader.read_single_file_stream_with_requested_ranges(
split,
file.as_ref().clone(),
data_fields.clone(),
deletion_vector,
row_ranges,
+ requested_row_ranges,
)
}
FieldSource::BlobBunch { bunch, .. } => {
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index 2647bff2..74c0c233 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -1572,6 +1572,15 @@ impl DataEvolutionPartialWriter {
(!write_set.omit_write_cols).then(||
write_set.write_columns.clone()),
)?
.with_file_index(write_set.file_index_options.clone());
+ let writer = if self
+ .write_sets
+ .iter()
+ .any(|set| set.kind == PartialFileKind::Vector)
+ {
+ writer.without_row_sidecar()
+ } else {
+ writer
+ };
self.writers.insert(key.clone(), writer);
}
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 82092a58..808f795b 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -330,6 +330,7 @@ impl DataFileReader {
)
});
split_file_offset += file_meta.row_count;
+ let sidecar_ranges = split_ranges.clone();
let selected_ranges = match file_index_result {
FileIndexResult::Remain => split_ranges,
FileIndexResult::Skip => Some(Vec::new()),
@@ -360,6 +361,7 @@ impl DataFileReader {
data_fields,
None,
row_selection,
+ sidecar_ranges,
)?;
while let Some(batch) = stream.next().await {
yield batch?;
@@ -436,6 +438,27 @@ impl DataFileReader {
data_fields: Option<Vec<DataField>>,
dv: Option<Arc<DeletionVector>>,
row_ranges: Option<Vec<RowRange>>,
+ ) -> crate::Result<ArrowRecordBatchStream> {
+ self.read_single_file_stream_with_requested_ranges(
+ split,
+ file_meta,
+ data_fields,
+ dv,
+ row_ranges.clone(),
+ row_ranges,
+ )
+ }
+
+ /// Physical selection may be narrowed by file indexes or a LIMIT, while
+ /// Java selects the ROW sidecar from the caller's original row ranges.
+ pub(super) fn read_single_file_stream_with_requested_ranges(
+ &self,
+ split: &DataSplit,
+ file_meta: DataFileMeta,
+ data_fields: Option<Vec<DataField>>,
+ dv: Option<Arc<DeletionVector>>,
+ row_ranges: Option<Vec<RowRange>>,
+ requested_ranges: Option<Vec<RowRange>>,
) -> crate::Result<ArrowRecordBatchStream> {
let local_ranges = row_ranges.as_ref().map(|ranges| {
to_local_row_ranges(
@@ -446,12 +469,20 @@ impl DataFileReader {
});
let row_selection =
merge_row_selection(file_meta.row_count, dv.as_deref(),
local_ranges.as_deref());
+ let sidecar_ranges = requested_ranges.as_ref().map(|ranges| {
+ to_local_row_ranges(
+ ranges,
+ file_meta.first_row_id.unwrap_or(0),
+ file_meta.row_count,
+ )
+ });
self.read_single_file_stream_with_selection(
split,
file_meta,
data_fields,
None,
row_selection,
+ sidecar_ranges,
)
}
@@ -482,6 +513,7 @@ impl DataFileReader {
data_fields,
Some(data_schema_fields),
row_selection,
+ local_ranges,
)
}
@@ -492,6 +524,7 @@ impl DataFileReader {
data_fields: Option<Vec<DataField>>,
data_schema_fields: Option<Vec<DataField>>,
row_selection: Option<Vec<RowRange>>,
+ sidecar_ranges: Option<Vec<RowRange>>,
) -> crate::Result<ArrowRecordBatchStream> {
if row_selection.as_ref().is_some_and(Vec::is_empty) {
return Ok(futures::stream::empty().boxed());
@@ -582,23 +615,6 @@ impl DataFileReader {
blob_parallelism,
mosaic_prefetch,
)?;
- let format_read_fields = configured_reader.read_fields;
- let format_reader = configured_reader.reader;
- // The decoded batch is described by `format_read_fields`, so map
- // `read_type` onto *that* list: its entries carry the types the
columns
- // actually come back as, which is what reconciling them needs.
- let (index_mapping, source_fields) = if data_fields.is_some()
- || file_meta.write_cols.is_some()
- || read_type.iter().any(contains_selected_map)
- {
- (
- create_index_mapping(&read_type, &format_read_fields),
- Some(format_read_fields.clone()),
- )
- } else {
- (None, None)
- };
-
// Remap predicates from table-level to file-level indices.
let file_predicates = if row_id_residual {
None
@@ -621,7 +637,40 @@ impl DataFileReader {
Ok(try_stream! {
let schema_open_start = read_timing.as_ref().map(|_|
Instant::now());
- let input_file = file_io.new_input(&path_to_read)?;
+ let options = crate::spec::CoreOptions::new(&table_options);
+ let sidecar_ranges = super::row_sidecar::supports_read_type(
+ &read_type, blob_as_descriptor || options.blob_as_descriptor(),
+ ).then_some(sidecar_ranges.as_deref()).flatten();
+ let (read_path, read_size) = super::row_sidecar::read_target(
+ &file_io, &file_meta, split.bucket_path(), sidecar_ranges,
&options,
+ ).await?;
+ let configured_reader = if read_path == path_to_read {
+ configured_reader
+ } else {
+ create_format_reader_with_budget(
+ &read_path, blob_as_descriptor,
+ FormatReadFields { data_schema: &data_schema_fields,
projected: &projected_read_fields },
+ &table_options, None, blob_parallelism, mosaic_prefetch,
+ )?
+ };
+ let format_read_fields = configured_reader.read_fields;
+ let format_reader = configured_reader.reader;
+ // The decoded batch is described by `format_read_fields`, so map
+ // `read_type` onto *that* list: its entries carry the types the
columns
+ // actually come back as, which is what reconciling them needs.
+ let (index_mapping, source_fields) = if data_fields.is_some()
+ || file_meta.write_cols.is_some()
+ || read_type.iter().any(contains_selected_map)
+ {
+ (
+ create_index_mapping(&read_type, &format_read_fields),
+ Some(format_read_fields.clone()),
+ )
+ } else {
+ (None, None)
+ };
+
+ let input_file = file_io.new_input(&read_path)?;
let open_start = read_timing.as_ref().map(|_| Instant::now());
let file_reader = input_file.reader().await?;
if let (Some(timing), Some(start)) = (read_timing.as_ref(),
open_start) {
@@ -634,7 +683,14 @@ impl DataFileReader {
}),
None => Box::new(file_reader),
};
- let is_parquet =
path_to_read.to_ascii_lowercase().ends_with(".parquet");
+ let is_parquet =
read_path.to_ascii_lowercase().ends_with(".parquet");
+ let target_schema = if read_path != path_to_read &&
read_path.ends_with(".row") {
+ let names = read_type.iter().filter(|field|
matches!(field.data_type(), DataType::Blob(_)))
+ .map(|field| field.name().to_string()).collect();
+ super::row_sidecar::mark_blob_data(&target_schema, &names)
+ } else {
+ target_schema
+ };
let selected_row_ids = selected_row_ids_for_read(
projects_row_id,
file_meta.first_row_id,
@@ -645,7 +701,7 @@ impl DataFileReader {
let mut batch_stream = format_reader.read_batch_stream(
file_reader,
- file_meta.file_size as u64,
+ read_size,
&format_read_fields,
file_predicates.as_ref(),
batch_size,
@@ -1477,7 +1533,8 @@ pub(super) fn insert_column_at(
if insert_index >= batch.num_columns() {
columns.push(column);
}
- RecordBatch::try_new(output_schema.clone(), columns).map_err(|e|
Error::UnexpectedError {
+ let schema = super::row_sidecar::propagate_blob_data(output_schema,
&batch);
+ RecordBatch::try_new(schema, columns).map_err(|e| Error::UnexpectedError {
message: format!("Failed to insert column into RecordBatch: {e}"),
source: Some(Box::new(e)),
})
diff --git a/crates/paimon/src/table/data_file_writer.rs
b/crates/paimon/src/table/data_file_writer.rs
index 20798816..17946951 100644
--- a/crates/paimon/src/table/data_file_writer.rs
+++ b/crates/paimon/src/table/data_file_writer.rs
@@ -24,6 +24,7 @@
use super::data_file_index_writer::{DataFileIndexWriter, FileIndexOptions};
use super::data_file_path_factory::{DataFilePath, DataFilePathFactory};
+use super::row_sidecar::RowSidecarWriter;
use crate::arrow::format::{
create_format_writer_factory, with_write_resources, FormatFileWriter,
FormatValueStats,
FormatWriterFactory,
@@ -75,6 +76,8 @@ pub(crate) struct DataFileWriter {
current_row_count: i64,
index_options: Option<Arc<FileIndexOptions>>,
current_index: Option<DataFileIndexWriter>,
+ row_sidecar_enabled: bool,
+ current_row_sidecar: Option<RowSidecarWriter>,
resources: Option<ResourceContext>,
/// Paths owned by this write until prepare_commit hands them to the
caller.
created_paths: Vec<String>,
@@ -108,7 +111,14 @@ impl DataFileWriter {
let data_file_prefix = CoreOptions::new(&format_options)
.data_file_prefix()
.to_string();
+ let options = CoreOptions::new(&format_options);
+ let row_sidecar_enabled = options.data_evolution_enabled()
+ && options.data_evolution_row_sidecar_enabled()?
+ && !file_format.eq_ignore_ascii_case("blob")
+ && !file_format.starts_with("vector.");
Ok(Self {
+ row_sidecar_enabled,
+ current_row_sidecar: None,
format_writer_factory: None,
file_io,
paths,
@@ -149,6 +159,13 @@ impl DataFileWriter {
self
}
+ /// Java's dedicated-format writer does not create auxiliary ROW files,
+ /// including for its normal columns.
+ pub(super) fn without_row_sidecar(mut self) -> Self {
+ self.row_sidecar_enabled = false;
+ self
+ }
+
pub(crate) fn with_target_file_row_num(mut self, rows: i64) -> Self {
debug_assert!(rows > 0);
self.target_file_row_num = rows;
@@ -167,7 +184,7 @@ impl DataFileWriter {
/// Write a RecordBatch. Rolls when either target size or row count is
reached.
pub(crate) async fn write(&mut self, batch: &RecordBatch) -> Result<()> {
let result = self.write_batch(batch).await;
- if self.index_options.is_some() && result.is_err() {
+ if (self.index_options.is_some() || self.row_sidecar_enabled) &&
result.is_err() {
self.abort().await;
}
result
@@ -184,6 +201,9 @@ impl DataFileWriter {
}
self.current_writer.as_mut().unwrap().write(batch).await?;
+ if let Some(sidecar) = self.current_row_sidecar.as_mut() {
+ sidecar.write(batch).await?;
+ }
if let Some(index) = &mut self.current_index {
index.write(batch)?;
}
@@ -246,7 +266,7 @@ impl DataFileWriter {
None => {
let factory = create_format_writer_factory(
&self.file_format,
- schema,
+ schema.clone(),
self.file_compression_zstd_level,
Some(self.file_io.clone()),
Some(&self.write_fields),
@@ -262,9 +282,36 @@ impl DataFileWriter {
.await?;
self.current_writer = Some(with_write_resources(writer,
self.resources.as_ref()));
self.current_index = index;
- self.current_file_name = Some(file_name);
+ self.current_file_name = Some(file_name.clone());
self.current_file_path = Some(location);
self.current_row_count = 0;
+ self.open_row_sidecar(&file_path, &file_name, schema)
+ .await?;
+ Ok(())
+ }
+
+ async fn open_row_sidecar(
+ &mut self,
+ path: &str,
+ file_name: &str,
+ schema: arrow_schema::SchemaRef,
+ ) -> Result<()> {
+ if self.row_sidecar_enabled {
+ let path = format!("{path}.row");
+ self.created_paths.push(path.clone());
+ self.current_row_sidecar = Some(
+ RowSidecarWriter::new(
+ &self.file_io,
+ &path,
+ file_name,
+ schema,
+ &self.write_fields,
+ &self.format_options,
+ self.resources.as_ref(),
+ )
+ .await?,
+ );
+ }
Ok(())
}
@@ -293,6 +340,7 @@ impl DataFileWriter {
) -> Option<impl std::future::Future<Output = Result<DataFileMeta>> + Send
+ 'static> {
let writer = self.current_writer.take()?;
let index = self.current_index.take();
+ let row_sidecar = self.current_row_sidecar.take();
let file_io = self.file_io.clone();
let location = self.current_file_path.take().unwrap();
let bucket_dir = location.parent().to_string();
@@ -316,7 +364,19 @@ impl DataFileWriter {
};
Some(async move {
- let write_result = writer.close().await?;
+ // Close both outputs even if an auxiliary close fails. All paths
+ // remain producer-owned until the whole prepare succeeds.
+ let sidecar_result = match row_sidecar {
+ Some(sidecar) => sidecar
+ .writer
+ .close()
+ .await
+ .map(|_| Some(sidecar.file_name)),
+ None => Ok(None),
+ };
+ let write_result = writer.close().await;
+ let sidecar_name = sidecar_result?;
+ let write_result = write_result?;
let mut meta = Self::build_meta(
file_name,
write_result.file_size as i64,
@@ -342,6 +402,9 @@ impl DataFileWriter {
meta.embedded_index = Some(bytes.to_vec());
}
}
+ if let Some(name) = sidecar_name {
+ meta.extra_files.push(name);
+ }
Ok(meta)
})
}
@@ -349,7 +412,7 @@ impl DataFileWriter {
/// Close the current writer and return all written file metadata.
pub(crate) async fn prepare_commit(&mut self) -> Result<Vec<DataFileMeta>>
{
let result = self.finish().await;
- if result.is_err() && self.index_options.is_some() {
+ if result.is_err() && (self.index_options.is_some() ||
self.row_sidecar_enabled) {
self.abort().await;
}
result
@@ -429,6 +492,9 @@ impl DataFileWriter {
}
pub(super) async fn abort(&mut self) {
+ if let Some(sidecar) = self.current_row_sidecar.take() {
+ let _ = sidecar.writer.close().await;
+ }
if let Some(writer) = self.current_writer.take() {
let _ = writer.close().await;
}
@@ -788,9 +854,154 @@ mod tests {
}
}
+ struct FailingSidecar {
+ inner: Box<dyn FormatFileWriter>,
+ fail_write: bool,
+ }
+
+ #[async_trait::async_trait]
+ impl FormatFileWriter for FailingSidecar {
+ async fn write(&mut self, batch: &RecordBatch) -> Result<()> {
+ if self.fail_write {
+ return Err(crate::Error::DataInvalid {
+ message: "injected sidecar write failure".into(),
+ source: None,
+ });
+ }
+ self.inner.write(batch).await
+ }
+ fn num_bytes(&self) -> usize {
+ self.inner.num_bytes()
+ }
+ fn in_progress_size(&self) -> usize {
+ self.inner.in_progress_size()
+ }
+ async fn flush(&mut self) -> Result<()> {
+ self.inner.flush().await
+ }
+ async fn close(self: Box<Self>) ->
Result<crate::arrow::format::FormatWriteResult> {
+ self.inner.close().await?;
+ Err(crate::Error::DataInvalid {
+ message: "injected sidecar close failure".into(),
+ source: None,
+ })
+ }
+ }
+
+ #[derive(Debug)]
+ struct SidecarFailureProvider {
+ operator: Operator,
+ fail_create: std::sync::atomic::AtomicBool,
+ }
+
+ #[async_trait::async_trait]
+ impl FileIOProvider for SidecarFailureProvider {
+ async fn create(&self, path: &str) -> Result<(Operator, String)> {
+ if path.ends_with(".row") &&
self.fail_create.load(std::sync::atomic::Ordering::Relaxed)
+ {
+ return Err(crate::Error::DataInvalid {
+ message: "injected sidecar create failure".into(),
+ source: None,
+ });
+ }
+ Ok((
+ self.operator.clone(),
+ path.trim_start_matches("memory:")
+ .trim_start_matches('/')
+ .into(),
+ ))
+ }
+ }
+
+ #[tokio::test]
+ async fn sidecar_failures_only_remove_current_producer_outputs() {
+ use std::sync::atomic::{AtomicBool, Ordering};
+ for failure in ["create", "write", "close"] {
+ let provider = Arc::new(SidecarFailureProvider {
+ operator:
Operator::from_config(opendal::services::MemoryConfig::default())
+ .unwrap(),
+ fail_create: AtomicBool::new(false),
+ });
+ let io = FileIOBuilder::new("unused")
+ .with_provider(provider.clone())
+ .build()
+ .unwrap();
+ let fields = vec![DataField::new(
+ 0,
+ "id".into(),
+ DataType::Int(IntType::new()),
+ )];
+ let mut writer = DataFileWriter::new(
+ io.clone(),
+ "memory:/sidecar-failure".into(),
+ String::new(),
+ 0,
+ 0,
+ i64::MAX,
+ "none".into(),
+ 0,
+ i64::MAX,
+ "parquet".into(),
+ fields,
+ HashMap::from([
+ ("data-evolution.enabled".into(), "true".into()),
+ ("data-evolution.row-sidecar.enabled".into(),
"true".into()),
+ ]),
+ Some(0),
+ None,
+ None,
+ )
+ .unwrap();
+ let batch = RecordBatch::try_from_iter([(
+ "id",
+ Arc::new(Int32Array::from(vec![1])) as arrow_array::ArrayRef,
+ )])
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ let prepared = writer.prepare_commit().await.unwrap();
+ let prior_paths = prepared[0].collect_files(writer.bucket_dir());
+ assert_eq!(prior_paths.len(), 2);
+ let error = if failure == "create" {
+ provider.fail_create.store(true, Ordering::Relaxed);
+ let error = writer.write(&batch).await.unwrap_err();
+ provider.fail_create.store(false, Ordering::Relaxed);
+ error
+ } else {
+ writer.write(&batch).await.unwrap();
+ let sidecar = writer.current_row_sidecar.take().unwrap();
+ let mut sidecar = sidecar;
+ sidecar.writer = Box::new(FailingSidecar {
+ inner: sidecar.writer,
+ fail_write: failure == "write",
+ });
+ writer.current_row_sidecar = Some(sidecar);
+ if failure == "write" {
+ writer.write(&batch).await.unwrap_err()
+ } else {
+ writer.prepare_commit().await.unwrap_err()
+ }
+ };
+ assert!(error
+ .to_string()
+ .contains(&format!("sidecar {failure} failure")));
+ writer.abort().await;
+ let mut actual: Vec<_> = io
+ .list_status_recursive(writer.bucket_dir())
+ .await
+ .unwrap()
+ .into_iter()
+ .map(|status| status.path)
+ .collect();
+ actual.sort();
+ let mut expected = prior_paths;
+ expected.sort();
+ assert_eq!(actual, expected, "{failure}");
+ }
+ }
+
#[tokio::test]
async fn grouped_prepare_failure_removes_successful_and_failed_outputs() {
- for external in [false, true] {
+ for (external, sidecar) in [(false, false), (true, false), (false,
true), (true, true)] {
let file_io = FileIOBuilder::new("memory").build().unwrap();
let mut writers = Vec::new();
for first_row_id in [0, 1] {
@@ -829,6 +1040,7 @@ mod tests {
None,
)
.unwrap();
+ writer.row_sidecar_enabled = sidecar;
let batch = RecordBatch::try_from_iter([(
"id",
Arc::new(Int32Array::from(vec![1])) as
arrow_array::ArrayRef,
@@ -854,7 +1066,7 @@ mod tests {
.await
.unwrap()
.iter()
- .all(|entry| !entry.path.ends_with(".parquet")));
+ .all(|entry| !entry.path.ends_with(".parquet") &&
!entry.path.ends_with(".row")));
}
}
}
diff --git a/crates/paimon/src/table/dedicated_format_file_writer.rs
b/crates/paimon/src/table/dedicated_format_file_writer.rs
index 901b7b0c..d02c4618 100644
--- a/crates/paimon/src/table/dedicated_format_file_writer.rs
+++ b/crates/paimon/src/table/dedicated_format_file_writer.rs
@@ -211,6 +211,13 @@ impl AppendDedicatedFormatFileWriter {
.with_path_factory(paths.clone())
.with_file_index(normal_index)
.with_target_file_row_num(target_file_row_num);
+ // A view-only schema also uses this adapter to translate references,
+ // but Java treats its physical writer as an ordinary inline writer.
+ let normal_writer = if !blob_writers.is_empty() ||
vector_writer.is_some() {
+ normal_writer.without_row_sidecar()
+ } else {
+ normal_writer
+ };
Ok(Self {
written_files: Vec::new(),
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 36e75d58..58134550 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -124,6 +124,7 @@ pub(crate) mod rest_env;
pub(crate) mod row_id_predicate;
mod row_kind_generator;
mod row_position_selection;
+mod row_sidecar;
mod scan_trace;
pub(crate) mod schema_manager;
pub(crate) mod snapshot_commit;
diff --git a/crates/paimon/src/table/row_sidecar.rs
b/crates/paimon/src/table/row_sidecar.rs
new file mode 100644
index 00000000..5cbdc953
--- /dev/null
+++ b/crates/paimon/src/table/row_sidecar.rs
@@ -0,0 +1,254 @@
+// 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.
+
+//! Aligned ROW files for normal Data Evolution data files, matching Java's
+//! RowSidecarAuxiliaryWriter and DataEvolutionSplitRead.readTarget.
+
+use super::blob_resolver::{resolve_descriptor_column, BlobReadLimiter};
+use crate::arrow::format::{create_format_writer_factory, with_write_resources,
FormatFileWriter};
+use crate::io::FileIO;
+use crate::resource::ResourceContext;
+use crate::spec::{CoreOptions, DataField, DataFileMeta};
+use crate::table::RowRange;
+use crate::Result;
+use arrow_array::{Array, LargeBinaryArray, RecordBatch};
+use arrow_schema::{Field, Schema, SchemaRef};
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+
+pub(super) struct RowSidecarWriter {
+ pub file_name: String,
+ pub writer: Box<dyn FormatFileWriter>,
+ io: FileIO,
+ descriptor_fields: HashSet<String>,
+ view_fields: HashSet<String>,
+}
+
+impl RowSidecarWriter {
+ pub async fn new(
+ io: &FileIO,
+ path: &str,
+ file_name: &str,
+ schema: SchemaRef,
+ fields: &[DataField],
+ options: &HashMap<String, String>,
+ resources: Option<&ResourceContext>,
+ ) -> Result<Self> {
+ // ROW stores logical values. Parquet's MAP / VARIANT storage layouts
+ // must not change the auxiliary file's schema or encoding.
+ let options = CoreOptions::new(options);
+ let factory = create_format_writer_factory(
+ "row",
+ schema,
+ options.file_compression_zstd_level(),
+ None,
+ Some(fields),
+ None,
+ None,
+ )?;
+ let writer = factory.create_writer(&io.new_output(path)?,
"zstd").await?;
+ Ok(Self {
+ file_name: format!("{file_name}.row"),
+ writer: with_write_resources(writer, resources),
+ io: io.clone(),
+ descriptor_fields: options.blob_descriptor_fields(),
+ view_fields: options.blob_view_fields(),
+ })
+ }
+
+ pub async fn write(&mut self, batch: &RecordBatch) -> Result<()> {
+ let mut descriptors = Vec::new();
+ for (index, field) in batch.schema().fields().iter().enumerate() {
+ if self.descriptor_fields.contains(field.name()) {
+ descriptors.push(index);
+ }
+ if self.view_fields.contains(field.name()) {
+ let values = blob_column(batch, index)?;
+ // Java BlobView.toData() rejects an unresolved view. The Arrow
+ // API supplies serialized references, without a resolved Blob.
+ if values.null_count() != values.len() {
+ return Err(crate::Error::DataInvalid {
+ message:
+ "ROW sidecars require resolved BLOB views;
BlobView is not resolved"
+ .into(),
+ source: None,
+ });
+ }
+ }
+ }
+ if descriptors.is_empty() {
+ return self.writer.write(batch).await;
+ }
+ let limiter = BlobReadLimiter::new();
+ // Like Java Blob.toData(), materialize one row at a time. A batch of
+ // small references must not retain all referenced payloads in memory.
+ for row in 0..batch.num_rows() {
+ let indices = arrow_array::UInt64Array::from(vec![row as u64]);
+ let data = arrow_select::take::take_record_batch(batch,
&indices).map_err(|error| {
+ crate::Error::DataInvalid {
+ message: format!("Failed to select ROW sidecar BLOB row:
{error}"),
+ source: Some(Box::new(error)),
+ }
+ })?;
+ let mut columns = data.columns().to_vec();
+ for &index in &descriptors {
+ // ROW stores the actual payload; Parquet keeps the reference.
+ columns[index] = Arc::new(
+ resolve_descriptor_column(
+ blob_column(&data, index)?,
+ &self.io,
+ limiter.clone(),
+ )
+ .await?,
+ );
+ }
+ let data = RecordBatch::try_new(data.schema(),
columns).map_err(|error| {
+ crate::Error::DataInvalid {
+ message: format!("Failed to build ROW sidecar BLOB row:
{error}"),
+ source: Some(Box::new(error)),
+ }
+ })?;
+ self.writer.write(&data).await?;
+ }
+ Ok(())
+ }
+}
+
+fn blob_column(batch: &RecordBatch, index: usize) -> Result<&LargeBinaryArray>
{
+ batch
+ .column(index)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!(
+ "ROW sidecar BLOB field '{}' requires LargeBinary",
+ batch.schema().field(index).name()
+ ),
+ source: None,
+ })
+}
+
+// Arrow represents both BlobData and BlobRef as LargeBinary. Preserve the
+// distinction only inside the Data Evolution pipeline; project_output removes
+// this marker before exposing batches. Raw bytes can themselves look exactly
+// like a descriptor, so magic-number detection cannot supply the distinction.
+const BLOB_DATA: &str = "paimon.row-sidecar.blob-data";
+
+pub(super) fn is_blob_data(field: &Field) -> bool {
+ field.metadata().contains_key(BLOB_DATA)
+}
+
+pub(super) fn mark_blob_data(schema: &SchemaRef, names: &HashSet<String>) ->
SchemaRef {
+ if names.is_empty() {
+ return schema.clone();
+ }
+ let fields: Vec<_> = schema
+ .fields()
+ .iter()
+ .map(|field| {
+ if !names.contains(field.name()) {
+ return field.clone();
+ }
+ let mut metadata = field.metadata().clone();
+ metadata.insert(BLOB_DATA.into(), "true".into());
+ Arc::new(field.as_ref().clone().with_metadata(metadata))
+ })
+ .collect();
+ Arc::new(Schema::new_with_metadata(fields, schema.metadata().clone()))
+}
+
+pub(super) fn propagate_blob_data(schema: &SchemaRef, batch: &RecordBatch) ->
SchemaRef {
+ let names = batch
+ .schema()
+ .fields()
+ .iter()
+ .filter(|field| is_blob_data(field))
+ .map(|field| field.name().clone())
+ .collect();
+ mark_blob_data(schema, &names)
+}
+
+/// Exactly one ROW extra file is required, as in Java. Do not guess between
+/// multiple sidecars or enable one merely because deletion vectors prune rows.
+fn selected_sidecar<'a>(
+ file: &'a DataFileMeta,
+ local_ranges: Option<&[RowRange]>,
+ options: &CoreOptions<'_>,
+) -> Result<Option<&'a str>> {
+ if !options.data_evolution_enabled() {
+ return Ok(None);
+ }
+ let mut names = file
+ .extra_files
+ .iter()
+ .filter(|name| name.ends_with(".row"));
+ let Some(name) = names.next() else {
+ return Ok(None);
+ };
+ if names.next().is_some() {
+ return Ok(None);
+ }
+ let max_rows = options.data_evolution_row_sidecar_max_selected_rows()?;
+ let max_ratio = options.data_evolution_row_sidecar_max_selection_ratio()?;
+ let Some(ranges) = local_ranges else {
+ return Ok(None);
+ };
+ if file.row_count <= 0
+ || file.file_name.ends_with(".blob")
+ || file.file_name.contains(".vector.")
+ {
+ return Ok(None);
+ }
+ // The caller clips, sorts and merges the requested ranges into local
+ // positions before intersecting them with indexes and deletion vectors.
+ let selected: i64 = ranges.iter().map(RowRange::count).sum();
+ Ok((selected > 0
+ && selected < file.row_count
+ && selected <= max_rows
+ && selected as f64 / file.row_count as f64 <= max_ratio)
+ .then_some(name.as_str()))
+}
+
+pub(super) fn supports_read_type(read_type: &[DataField], blob_as_descriptor:
bool) -> bool {
+ // BlobData in ROW has no URI. A caller requesting descriptors must keep
+ // the primary file, rather than receive payload bytes in place of refs.
+ !blob_as_descriptor
+ || !read_type
+ .iter()
+ .any(|field| matches!(field.data_type(),
crate::spec::DataType::Blob(_)))
+}
+
+pub(super) async fn read_target(
+ io: &FileIO,
+ file: &DataFileMeta,
+ bucket_path: &str,
+ local_ranges: Option<&[RowRange]>,
+ options: &CoreOptions<'_>,
+) -> Result<(String, u64)> {
+ if let Some(name) = selected_sidecar(file, local_ranges, options)? {
+ let path = file.aligned_file_path(bucket_path, name);
+ match io.get_status(&path).await {
+ Ok(status) => return Ok((path, status.size)),
+ Err(error) => {
+ if !options.scan_ignore_lost_file() {
+ return Err(error);
+ }
+ }
+ }
+ }
+ Ok((file.data_file_path(bucket_path), file.file_size as u64))
+}
diff --git a/crates/paimon/tests/data_evolution_row_sidecar_test.rs
b/crates/paimon/tests/data_evolution_row_sidecar_test.rs
new file mode 100644
index 00000000..c8814e3b
--- /dev/null
+++ b/crates/paimon/tests/data_evolution_row_sidecar_test.rs
@@ -0,0 +1,815 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+
+use arrow_array::{ArrayRef, Int32Array, Int64Array, RecordBatch, StringArray};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+use futures::TryStreamExt;
+use paimon::io::FileIO;
+use paimon::spec::{DataType, IntType, Schema, TableSchema, VarCharType};
+use paimon::table::{CommitMessage, RowRange, Table};
+use std::collections::HashMap;
+use std::sync::Arc;
+
+#[tokio::test]
+async fn append_registers_aligned_row_sidecars_in_data_file_metadata() {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::VarChar(VarCharType::string_type()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option("data-evolution.row-sidecar.enabled", "true")
+ .option("target-file-row-num", "2")
+ .build()
+ .unwrap();
+ let (io, table) = memory_table("memory:/de_row_sidecar",
TableSchema::new(0, &schema));
+ setup_dirs(&io, table.location()).await;
+ persist_table_schema(&io, table.location(), table.schema()).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ for ids in [vec![1, 2], vec![3]] {
+ let values: Vec<String> = ids.iter().map(|id|
format!("value-{id}")).collect();
+ writer
+ .write_arrow_batch(
+ &RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(ids)) as
arrow_array::ArrayRef,
+ ),
+ (
+ "value",
+ Arc::new(StringArray::from(values)) as
arrow_array::ArrayRef,
+ ),
+ ])
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ }
+ let messages = writer.prepare_commit().await.unwrap();
+ let files: Vec<_> = messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .collect();
+ assert_eq!(files.len(), 2);
+ for file in &files {
+ assert_eq!(file.extra_files, vec![format!("{}.row", file.file_name)]);
+ for path in file.collect_files(&format!("{}/bucket-0",
table.location())) {
+ assert!(io.exists(&path).await.unwrap(), "{path}");
+ }
+ }
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .next_row_id(),
+ Some(3)
+ );
+}
+
+async fn table(options: &[(&str, &str)]) -> Table {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::VarChar(VarCharType::string_type()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option("deletion-vectors.enabled", "true")
+ .option("data-evolution.row-sidecar.enabled", "true");
+ for (name, value) in options {
+ schema = schema.option(*name, *value);
+ }
+ let schema = TableSchema::new(0, &schema.build().unwrap());
+ let (io, table) = memory_table("memory:/row_sidecar", schema);
+ setup_dirs(&io, table.location()).await;
+ persist_table_schema(&io, table.location(), table.schema()).await;
+ table
+}
+
+fn batch(start: i32, count: i32) -> RecordBatch {
+ RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from_iter_values(start..start + count)) as
ArrayRef,
+ ),
+ (
+ "value",
+ Arc::new(StringArray::from_iter_values(
+ (start..start + count).map(|id| format!("v{id}")),
+ )) as ArrayRef,
+ ),
+ ])
+ .unwrap()
+}
+
+async fn append(table: &Table, start: i32, count: i32) -> Vec<CommitMessage> {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&batch(start, count))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ // Closing a producer after prepare must retain both handed-off outputs.
+ writer.close().await;
+ builder.new_commit().commit(messages.clone()).await.unwrap();
+ messages
+}
+
+async fn read(table: &Table, ranges: Option<Vec<RowRange>>) ->
paimon::Result<Vec<RecordBatch>> {
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["id", "value", "_ROW_ID"])?;
+ if let Some(ranges) = ranges {
+ builder.with_row_ranges(ranges);
+ }
+ let plan = builder.new_scan().plan().await?;
+ builder
+ .new_read()?
+ .to_arrow(plan.splits())?
+ .try_collect()
+ .await
+}
+
+fn rows(batches: &[RecordBatch]) -> Vec<(i32, String, i64)> {
+ batches
+ .iter()
+ .flat_map(|batch| {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let values = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let row_ids = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ (0..batch.num_rows())
+ .map(|row| (ids.value(row), values.value(row).into(),
row_ids.value(row)))
+ .collect::<Vec<_>>()
+ })
+ .collect()
+}
+
+async fn remove_primary(io: &FileIO, table: &Table, messages:
&[CommitMessage]) {
+ for file in messages.iter().flat_map(|message| &message.new_files) {
+ io.delete_file(&file.data_file_path(&format!("{}/bucket-0",
table.location())))
+ .await
+ .unwrap();
+ }
+}
+
+#[tokio::test]
+async fn
sparse_ranges_use_sidecar_with_nonzero_ids_external_paths_and_read_option_disabled()
{
+ for external in [false, true] {
+ let options = if external {
+ vec![("data-file.external-paths", "memory:/external%20sidecars")]
+ } else {
+ vec![]
+ };
+ let table = table(&options).await;
+ append(&table, -20, 20).await;
+ let messages = append(&table, 0, 100).await;
+ remove_primary(table.file_io(), &table, &messages).await;
+ let read_table = table.copy_with_options(HashMap::from([(
+ "data-evolution.row-sidecar.enabled".into(),
+ "false".into(),
+ )]));
+ let actual = read(
+ &read_table,
+ Some(vec![
+ RowRange::new(25, 26),
+ RowRange::new(26, 27),
+ RowRange::new(150, 200),
+ ]),
+ )
+ .await
+ .unwrap();
+ assert_eq!(
+ rows(&actual),
+ vec![
+ (5, "v5".into(), 25),
+ (6, "v6".into(), 26),
+ (7, "v7".into(), 27)
+ ]
+ );
+ }
+}
+
+#[tokio::test]
+async fn selection_thresholds_are_inclusive_but_full_reads_keep_primary() {
+ for (count, max_rows, max_ratio, expected_sidecar) in [
+ (5, "5", "0.05", true),
+ (6, "4096", "0.05", false),
+ (5, "4", "0.05", false),
+ (6, "6", "0.06", true),
+ (100, "4096", "1", false),
+ ] {
+ let table = table(&[
+ ("data-evolution.row-sidecar.max-selected-rows", max_rows),
+ ("data-evolution.row-sidecar.max-selection-ratio", max_ratio),
+ ])
+ .await;
+ let messages = append(&table, 0, 100).await;
+ remove_primary(table.file_io(), &table, &messages).await;
+ let result = read(&table, Some(vec![RowRange::new(0, count -
1)])).await;
+ if expected_sidecar {
+ assert_eq!(rows(&result.unwrap()).len(), count as usize);
+ } else {
+ assert!(
+ result.is_err(),
+ "count={count}, max rows={max_rows}, max ratio={max_ratio}"
+ );
+ }
+ assert!(read(&table, None).await.is_err());
+ }
+}
+
+#[tokio::test]
+async fn missing_sidecar_only_falls_back_when_ignore_lost_file_is_enabled() {
+ let table = table(&[]).await;
+ let messages = append(&table, 0, 100).await;
+ let file = &messages[0].new_files[0];
+ let sidecar = file.aligned_file_path(
+ &format!("{}/bucket-0", table.location()),
+ &file.extra_files[0],
+ );
+ table.file_io().delete_file(&sidecar).await.unwrap();
+ let strict = table.copy_with_options(HashMap::from([(
+ "scan.ignore-lost-files".into(),
+ "false".into(),
+ )]));
+ for reader_table in [&table, &strict] {
+ let error = read(reader_table, Some(vec![RowRange::new(5, 5)]))
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains(".row"), "{error}");
+ }
+ let tolerant = table.copy_with_options(HashMap::from([(
+ "scan.ignore-lost-files".into(),
+ "true".into(),
+ )]));
+ assert_eq!(
+ rows(
+ &read(&tolerant, Some(vec![RowRange::new(5, 5)]))
+ .await
+ .unwrap()
+ ),
+ vec![(5, "v5".into(), 5)]
+ );
+ // Whole-file reads never require the auxiliary file.
+ assert_eq!(rows(&read(&table, None).await.unwrap()).len(), 100);
+}
+
+#[tokio::test]
+async fn corrupt_sidecar_is_not_hidden_by_ignore_lost_file() {
+ let table = table(&[("scan.ignore-lost-files", "true")]).await;
+ let messages = append(&table, 0, 100).await;
+ let file = &messages[0].new_files[0];
+ let path = file.aligned_file_path(
+ &format!("{}/bucket-0", table.location()),
+ &file.extra_files[0],
+ );
+ table
+ .file_io()
+ .new_output(&path)
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"corrupt"))
+ .await
+ .unwrap();
+ assert!(read(&table, Some(vec![RowRange::new(5, 5)])).await.is_err());
+}
+
+#[tokio::test]
+async fn
updated_columns_and_deletion_vectors_are_applied_when_reading_sidecars() {
+ for incremental in [false, true] {
+ let table = table(&[]).await;
+ let initial = append(&table, 0, 100).await;
+ let input = RecordBatch::try_from_iter([
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from(vec![5, 7])) as ArrayRef,
+ ),
+ (
+ "value",
+ Arc::new(StringArray::from(vec!["changed-5", "changed-7"])) as
ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let update = table.new_write_builder().new_update().unwrap();
+ let messages = if incremental {
+ update
+ .new_update_by_row_id()
+ .await
+ .unwrap()
+ .update_columns(vec![input], vec!["value".into()])
+ .await
+ .unwrap()
+ } else {
+ update
+ .update_by_arrow_with_row_id(vec![input])
+ .await
+ .unwrap()
+ };
+ assert!(messages
+ .iter()
+ .flat_map(|m| &m.new_files)
+ .all(|file| file.extra_files == vec![format!("{}.row",
file.file_name)]));
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages.clone())
+ .await
+ .unwrap();
+ let deletions = update.delete_by_row_id(vec![6]).await.unwrap();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(deletions)
+ .await
+ .unwrap();
+ remove_primary(table.file_io(), &table, &initial).await;
+ remove_primary(table.file_io(), &table, &messages).await;
+ assert_eq!(
+ rows(&read(&table, Some(vec![RowRange::new(5,
7)])).await.unwrap()),
+ vec![(5, "changed-5".into(), 5), (7, "changed-7".into(), 7)]
+ );
+ }
+}
+
+#[tokio::test]
+async fn
explicit_abort_deletes_sidecar_while_writer_close_keeps_prepared_files() {
+ let table = table(&[]).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&batch(0, 5)).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let paths = messages[0].new_files[0].collect_files(&format!("{}/bucket-0",
table.location()));
+ assert_eq!(paths.len(), 2);
+ writer.close().await;
+ for path in &paths {
+ assert!(table.file_io().exists(path).await.unwrap());
+ }
+ builder.new_commit().abort(&messages).await.unwrap();
+ for path in &paths {
+ assert!(!table.file_io().exists(path).await.unwrap());
+ }
+ // A producer close before prepare owns and removes all its current files.
+ writer.write_arrow_batch(&batch(5, 5)).await.unwrap();
+ writer.close().await;
+ assert!(table
+ .file_io()
+ .list_status_recursive(&format!("{}/bucket-0", table.location()))
+ .await
+ .unwrap()
+ .is_empty());
+}
+
+#[tokio::test]
+async fn file_index_and_limit_do_not_turn_full_reads_into_sidecar_reads() {
+ use paimon::spec::{Datum, PredicateBuilder};
+ let table = table(&[("file-index.bitmap.columns", "id")]).await;
+ let messages = append(&table, 0, 100).await;
+ for file in messages.iter().flat_map(|m| &m.new_files) {
+ let sidecar = file
+ .extra_files
+ .iter()
+ .find(|name| name.ends_with(".row"))
+ .unwrap();
+ table
+ .file_io()
+ .delete_file(
+ &file.aligned_file_path(&format!("{}/bucket-0",
table.location()), sidecar),
+ )
+ .await
+ .unwrap();
+ }
+ // A one-row bitmap selection must still decode the existing primary file.
+ for requested_ranges in [None, Some(vec![RowRange::new(0, 99)])] {
+ let mut builder = table.new_read_builder();
+ builder.with_filter(
+ PredicateBuilder::new(table.schema().fields())
+ .equal("id", Datum::Int(5))
+ .unwrap(),
+ );
+ if let Some(ranges) = requested_ranges {
+ builder.with_row_ranges(ranges);
+ }
+ let plan = builder.new_scan().plan().await.unwrap();
+ let actual: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(actual.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ }
+ let mut builder = table.new_read_builder();
+ builder.with_limit(1);
+ let plan = builder.new_scan().plan().await.unwrap();
+ let actual: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(actual.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ let update = table.new_write_builder().new_update().unwrap();
+ let input = RecordBatch::try_from_iter([
+ ("_ROW_ID", Arc::new(Int64Array::from(vec![0])) as ArrayRef),
+ (
+ "value",
+ Arc::new(StringArray::from(vec!["changed"])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let messages = update
+ .update_by_arrow_with_row_id(vec![input])
+ .await
+ .unwrap();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages.clone())
+ .await
+ .unwrap();
+ for file in messages.iter().flat_map(|m| &m.new_files) {
+ let name = file
+ .extra_files
+ .iter()
+ .find(|name| name.ends_with(".row"))
+ .unwrap();
+ table
+ .file_io()
+ .delete_file(&file.aligned_file_path(&format!("{}/bucket-0",
table.location()), name))
+ .await
+ .unwrap();
+ }
+ let mut builder = table.new_read_builder();
+ builder.with_limit(1);
+ let plan = builder.new_scan().plan().await.unwrap();
+ let actual: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(actual[0].num_rows(), 1);
+ assert_eq!(
+ actual[0]
+ .column(1)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .value(0),
+ "changed"
+ );
+}
+
+#[tokio::test]
+async fn inline_vector_sidecar_preserves_values_and_parent_nulls() {
+ use arrow_array::{
+ builder::{FixedSizeListBuilder, Float32Builder},
+ Array, FixedSizeListArray, Float32Array,
+ };
+ use paimon::spec::{FloatType, VectorType};
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "embedding",
+ DataType::Vector(VectorType::new(3,
DataType::Float(FloatType::new())).unwrap()),
+ )
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option("data-evolution.row-sidecar.enabled", "true")
+ .build()
+ .unwrap();
+ let (io, table) = memory_table("memory:/inline_vector_row",
TableSchema::new(0, &schema));
+ setup_dirs(&io, table.location()).await;
+ persist_table_schema(&io, table.location(), table.schema()).await;
+ let mut values = FixedSizeListBuilder::new(Float32Builder::new(),
3).with_field(Arc::new(
+ arrow_schema::Field::new("element", arrow_schema::DataType::Float32,
true),
+ ));
+ for id in 0..100 {
+ for dimension in 0..3 {
+ values.values().append_value((id * 3 + dimension) as f32);
+ }
+ values.append(id != 6);
+ }
+ let data = RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from_iter_values(0..100)) as ArrayRef,
+ ),
+ ("embedding", Arc::new(values.finish()) as ArrayRef),
+ ])
+ .unwrap();
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&data).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(messages[0].new_files[0].extra_files.len(), 1);
+ builder.new_commit().commit(messages.clone()).await.unwrap();
+ remove_primary(&io, &table, &messages).await;
+ let mut builder = table.new_read_builder();
+ builder.with_row_ranges(vec![RowRange::new(5, 6)]);
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let vectors = batches[0]
+ .column(1)
+ .as_any()
+ .downcast_ref::<FixedSizeListArray>()
+ .unwrap();
+ assert_eq!(vectors.len(), 2);
+ assert!(vectors.is_null(1));
+ let values = vectors.value(0);
+ assert_eq!(
+ values
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .unwrap()
+ .values()
+ .as_ref(),
+ &[15.0, 16.0, 17.0]
+ );
+}
+
+#[tokio::test]
+async fn inline_blob_views_generate_normal_sidecars() {
+ use arrow_array::LargeBinaryArray;
+ use paimon::catalog::Identifier;
+ use paimon::spec::{BlobType, BlobViewStruct};
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", DataType::Blob(BlobType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option("data-evolution.row-sidecar.enabled", "true")
+ .option("blob-view-field", "payload")
+ .build()
+ .unwrap();
+ let (io, table) = memory_table("memory:/blob_view_row",
TableSchema::new(0, &schema));
+ setup_dirs(&io, table.location()).await;
+ persist_table_schema(&io, table.location(), table.schema()).await;
+ let value = BlobViewStruct::new(Identifier::new("db", "source"), 1, 2)
+ .serialize()
+ .unwrap();
+ let data = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![1])) as ArrayRef),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from_iter([None::<&[u8]>])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ writer.write_arrow_batch(&data).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let files: Vec<_> = messages.iter().flat_map(|m| &m.new_files).collect();
+ assert_eq!(files.len(), 1);
+ assert_eq!(
+ files[0].extra_files,
+ vec![format!("{}.row", files[0].file_name)]
+ );
+ let unresolved = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![2])) as ArrayRef),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from_iter([Some(value.as_slice())])) as
ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let error = writer.write_arrow_batch(&unresolved).await.unwrap_err();
+ assert!(error.to_string().contains("BlobView is not resolved"));
+}
+
+#[test]
+fn invalid_row_sidecar_selection_options_are_rejected() {
+ use paimon::spec::CoreOptions;
+ let options = HashMap::from([(
+ "data-evolution.row-sidecar.enabled".into(),
+ "invalid".into(),
+ )]);
+ assert!(CoreOptions::new(&options)
+ .data_evolution_row_sidecar_enabled()
+ .is_err());
+ for value in ["0", "-1", "invalid"] {
+ let options = HashMap::from([(
+ "data-evolution.row-sidecar.max-selected-rows".into(),
+ value.into(),
+ )]);
+ assert!(CoreOptions::new(&options)
+ .data_evolution_row_sidecar_max_selected_rows()
+ .is_err());
+ }
+ for value in ["0", "-0.1", "1.1", "invalid"] {
+ let options = HashMap::from([(
+ "data-evolution.row-sidecar.max-selection-ratio".into(),
+ value.into(),
+ )]);
+ assert!(CoreOptions::new(&options)
+ .data_evolution_row_sidecar_max_selection_ratio()
+ .is_err());
+ }
+}
+
+#[tokio::test]
+async fn automatic_row_block_flush_releases_memory_reservations() {
+ use paimon::resource::ResourceContext;
+ let table = table(&[("write.parquet-buffer-size", "1b")]).await;
+ let resources = ResourceContext::builder()
+ .memory_limit(512 * 1024)
+ .build()
+ .unwrap();
+ let builder = table.new_write_builder().with_resources(resources.clone());
+ let mut writer = builder.new_write().unwrap();
+ for index in 0..100 {
+ writer
+ .write_arrow_batch(&batch(index * 1000, 1000))
+ .await
+ .unwrap();
+ assert!(resources.metrics().reserved_memory_bytes < 256 * 1024);
+ }
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(resources.metrics().reserved_memory_bytes, 0);
+ assert_eq!(messages[0].new_files[0].row_count, 100000);
+}
+
+#[tokio::test]
+async fn
inline_blob_sidecars_materialize_references_and_keep_raw_payloads_after_merging()
{
+ use arrow_array::{Array, LargeBinaryArray};
+ use paimon::spec::{BlobDescriptor, BlobType};
+ for option in ["blob-descriptor-field", "blob.stored-descriptor-fields"] {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", DataType::Blob(BlobType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option("data-evolution.row-sidecar.enabled", "true")
+ .option(option, "payload")
+ .build()
+ .unwrap();
+ let (io, table) = memory_table("memory:/inline_blob_sidecar",
TableSchema::new(0, &schema));
+ setup_dirs(&io, table.location()).await;
+ persist_table_schema(&io, table.location(), table.schema()).await;
+ // The actual payload itself is a valid descriptor. Format provenance,
+ // rather than inspecting magic bytes, must distinguish BlobData.
+ let payload = BlobDescriptor::new("memory:/must-not-read".into(), 0,
1).serialize();
+ let source = [
+ b"prefix".as_slice(),
+ payload.as_slice(),
+ b"suffix".as_slice(),
+ ]
+ .concat();
+ io.new_output("memory:/source")
+ .unwrap()
+ .write(bytes::Bytes::from(source))
+ .await
+ .unwrap();
+ let reference =
+ BlobDescriptor::new("memory:/source".into(), 6, payload.len() as
i64).serialize();
+ let input = RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from_iter_values(0..100)) as ArrayRef,
+ ),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from_iter(
+ (0..100).map(|id| (id !=
6).then_some(reference.as_slice())),
+ )) as ArrayRef,
+ ),
+ (
+ "value",
+ Arc::new(Int32Array::from_iter_values(0..100)) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&input).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ builder.new_commit().commit(messages.clone()).await.unwrap();
+ let descriptor_table = table.copy_with_options(HashMap::from([(
+ "blob-as-descriptor".into(),
+ "true".into(),
+ )]));
+ for ranges in [None, Some(vec![RowRange::new(5, 5)])] {
+ let mut read = descriptor_table.new_read_builder();
+ read.with_projection(&["payload"]).unwrap();
+ if let Some(ranges) = ranges {
+ read.with_row_ranges(ranges);
+ }
+ let plan = read.new_scan().plan().await.unwrap();
+ let actual: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let blobs = actual[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap();
+ assert_eq!(blobs.value(0), reference);
+ }
+ let update = table.new_write_builder().new_update().unwrap();
+ let changed = RecordBatch::try_from_iter([
+ ("_ROW_ID", Arc::new(Int64Array::from(vec![5])) as ArrayRef),
+ ("value", Arc::new(Int32Array::from(vec![999])) as ArrayRef),
+ ])
+ .unwrap();
+ let updates = update
+ .update_by_arrow_with_row_id(vec![changed])
+ .await
+ .unwrap();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(updates.clone())
+ .await
+ .unwrap();
+ io.delete_file("memory:/source").await.unwrap();
+ remove_primary(&io, &table, &messages).await;
+ remove_primary(&io, &table, &updates).await;
+ for with_row_id in [false, true] {
+ let mut read = table.new_read_builder();
+ read.with_projection(if with_row_id {
+ &["id", "payload", "value", "_ROW_ID"]
+ } else {
+ &["id", "payload", "value"]
+ })
+ .unwrap();
+ read.with_row_ranges(vec![RowRange::new(5, 6)]);
+ let plan = read.new_scan().plan().await.unwrap();
+ let actual: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(actual.len(), 1);
+ let blobs = actual[0]
+ .column(1)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap();
+ assert_eq!(blobs.value(0), payload);
+ assert!(blobs.is_null(1));
+ let values = actual[0]
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(values.values().as_ref(), &[999, 6]);
+ assert!(actual[0]
+ .schema()
+ .field(1)
+ .metadata()
+ .keys()
+ .all(|key| !key.contains("blob-data")));
+ }
+ }
+}
