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 f107ce1f feat(core): support sparse raw Blob column updates (#1081)
f107ce1f is described below
commit f107ce1fbd5fef13975dc4176f69b00062e0dfff
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 9 10:00:59 2026 +0800
feat(core): support sparse raw Blob column updates (#1081)
---
.../python/python/pypaimon_rust/datafusion.pyi | 1 +
bindings/python/src/write.rs | 17 +-
crates/integrations/datafusion/tests/blob_tests.rs | 187 ++++-
crates/paimon/src/arrow/format/blob/writer.rs | 114 ++-
.../src/arrow/format/blob/writer_consumer_tests.rs | 60 +-
crates/paimon/src/table/data_evolution_blob.rs | 253 ++++++
crates/paimon/src/table/data_evolution_writer.rs | 361 +++++++--
crates/paimon/src/table/data_file_writer.rs | 28 +-
crates/paimon/src/table/table_update_by_row_id.rs | 13 +
crates/paimon/tests/blob_collection_write_test.rs | 14 -
crates/paimon/tests/blob_update_test.rs | 890 +++++++++++++++++++++
11 files changed, 1808 insertions(+), 130 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 65f7090f..af2b9f3b 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -320,6 +320,7 @@ class BatchTableUpdate:
...
class TableUpdateByRowId:
+ def _with_blob_uri_reader_factory(self, factory: Optional[Any]) -> None:
...
def update_columns(self, data: pyarrow.Table, column_names: List[str]) ->
List[CommitMessage]:
"""Stage one input table; return the cumulative messages to commit
once."""
...
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index ea65b1be..eec73110 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -877,6 +877,21 @@ pub struct PyTableUpdateByRowId {
#[pymethods]
impl PyTableUpdateByRowId {
+ fn _with_blob_uri_reader_factory(
+ &mut self,
+ py: Python<'_>,
+ factory: Option<Py<PyAny>>,
+ ) -> PyResult<()> {
+ let factory = factory
+ .map(|factory| {
+ PythonUriReaderFactory::new(factory, py)
+ .map(|factory| Arc::new(factory) as Arc<dyn
paimon::io::UriReaderFactory>)
+ })
+ .transpose()?;
+ self.inner.with_blob_uri_reader_factory(factory);
+ Ok(())
+ }
+
fn update_columns(
&mut self,
py: Python<'_>,
@@ -886,7 +901,7 @@ impl PyTableUpdateByRowId {
let batches = arrow_table_batches(data)?;
let messages = py
.detach(|| runtime().block_on(self.inner.update_columns(batches,
column_names)))
- .map_err(to_py_err)?;
+ .map_err(to_callback_error)?;
Ok(wrap_messages(
messages,
&self.table_location,
diff --git a/crates/integrations/datafusion/tests/blob_tests.rs
b/crates/integrations/datafusion/tests/blob_tests.rs
index 8b2a9aed..3c5f44a3 100644
--- a/crates/integrations/datafusion/tests/blob_tests.rs
+++ b/crates/integrations/datafusion/tests/blob_tests.rs
@@ -21,7 +21,7 @@
mod common;
-use arrow_array::{Array, Int32Array, LargeBinaryArray, RecordBatch};
+use arrow_array::{Array, Int32Array, LargeBinaryArray, ListArray, RecordBatch};
use common::{assert_sql_error, create_sql_context, create_test_env, exec,
string_value};
use paimon::catalog::Identifier;
use paimon::spec::{BlobDescriptor, BlobViewStruct};
@@ -340,37 +340,200 @@ async fn
test_merge_into_updates_non_blob_on_raw_blob_table() {
);
}
-/// Reference: BlobTestBase "Blob: merge-into rejects updating raw-data BLOB
column"
+/// Reference: Java Spark BlobUpdateTestBase raw-data BLOB and NULL updates.
#[tokio::test]
-async fn test_merge_into_rejects_raw_blob_update() {
+async fn test_merge_into_updates_raw_blob_and_preserves_unmatched_rows() {
let (_tmp, sql_context) = setup(BLOB_TABLE_DDL).await;
exec(
&sql_context,
- "INSERT INTO paimon.test_db.t (id, name, picture) VALUES (1, 'Alice',
X'4141')",
+ "INSERT INTO paimon.test_db.t (id, name, picture) VALUES \
+ (1, 'Alice', X'4141'), (2, 'Bob', X'4242'), (3, 'Carol', X'4343'), \
+ (4, 'Dave', NULL), (5, 'Eve', NULL)",
)
.await;
exec(
&sql_context,
- "CREATE TEMPORARY TABLE paimon.test_db.src AS SELECT * FROM (VALUES
(1, X'4242')) AS t(id, picture)",
+ "CREATE TEMPORARY TABLE paimon.test_db.src AS SELECT * FROM \
+ (VALUES (1, X'4444'), (3, CAST(NULL AS BYTEA)), (5, X'4545'), (99,
X'4646')) \
+ AS t(id, picture)",
)
.await;
- let result = sql_context
- .sql(
+ exec(
+ &sql_context,
+ "MERGE INTO paimon.test_db.t t \
+ USING paimon.test_db.src s ON t.id = s.id \
+ WHEN MATCHED THEN UPDATE SET picture = s.picture",
+ )
+ .await;
+ assert_eq!(
+ query_id_name_picture(
+ &sql_context,
+ "SELECT id, name, picture FROM paimon.test_db.t"
+ )
+ .await,
+ vec![
+ (1, "Alice".into(), Some(b"DD".to_vec())),
+ (2, "Bob".into(), Some(b"BB".to_vec())),
+ (3, "Carol".into(), None),
+ (4, "Dave".into(), None),
+ (5, "Eve".into(), Some(b"EE".to_vec())),
+ ]
+ );
+}
+
+#[tokio::test]
+async fn test_update_raw_blob_preserves_unmatched_rows() {
+ let (_tmp, sql_context) = setup(BLOB_TABLE_DDL).await;
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.t (id, name, picture) VALUES \
+ (1, 'Alice', X'4141'), (2, 'Bob', X'4242'), (3, 'Carol', X'4343')",
+ )
+ .await;
+
+ exec(
+ &sql_context,
+ "UPDATE paimon.test_db.t SET picture = X'4444', name = 'Updated' WHERE
id = 2",
+ )
+ .await;
+ exec(
+ &sql_context,
+ "UPDATE paimon.test_db.t SET picture = NULL WHERE id = 3",
+ )
+ .await;
+ exec(
+ &sql_context,
+ "UPDATE paimon.test_db.t SET picture = X'4545' WHERE id = 99",
+ )
+ .await;
+ assert_eq!(
+ query_id_name_picture(
+ &sql_context,
+ "SELECT id, name, picture FROM paimon.test_db.t"
+ )
+ .await,
+ vec![
+ (1, "Alice".into(), Some(b"AA".to_vec())),
+ (2, "Updated".into(), Some(b"DD".to_vec())),
+ (3, "Carol".into(), None),
+ ]
+ );
+}
+
+#[tokio::test]
+async fn test_merge_into_updates_raw_blob_collections() {
+ // Java Spark BlobUpdateTestBase also covers ARRAY<BLOB> and MAP<X, BLOB>.
+ for (blob_type, original, keep, updated, empty, projection) in [
+ (
+ "ARRAY<BLOB>",
+ "[X'4141']",
+ "[X'4242']",
+ "[X'4343', CAST(NULL AS BYTEA)]",
+ "CAST([] AS BYTEA[])",
+ "picture",
+ ),
+ (
+ "MAP(STRING, BLOB)",
+ "map(['old'], [X'4141'])",
+ "map(['keep'], [X'4242'])",
+ "map(['new', 'missing'], [X'4343', CAST(NULL AS BYTEA)])",
+ "map(CAST([] AS STRING[]), CAST([] AS BYTEA[]))",
+ "map_values(picture)",
+ ),
+ ] {
+ let (_tmp, sql_context) = setup(&format!(
+ "CREATE TABLE paimon.test_db.t (id INT, picture {blob_type}) WITH
(\
+ 'data-evolution.enabled' = 'true', 'row-tracking.enabled' =
'true')"
+ ))
+ .await;
+ exec(
+ &sql_context,
+ &format!(
+ "INSERT INTO paimon.test_db.t (id, picture) VALUES \
+ (1, {original}), (2, {keep}), (3, NULL), (4, {original})"
+ ),
+ )
+ .await;
+ exec(
+ &sql_context,
+ &format!(
+ "CREATE TABLE paimon.test_db.src (id INT, picture {blob_type})
WITH (\
+ 'data-evolution.enabled' = 'true', 'row-tracking.enabled'
= 'true')"
+ ),
+ )
+ .await;
+ exec(
+ &sql_context,
+ &format!(
+ "INSERT INTO paimon.test_db.src (id, picture) VALUES \
+ (1, {updated}), (3, {empty}), (4, NULL), (99, {original})"
+ ),
+ )
+ .await;
+ exec(
+ &sql_context,
"MERGE INTO paimon.test_db.t t \
USING paimon.test_db.src s ON t.id = s.id \
WHEN MATCHED THEN UPDATE SET picture = s.picture",
)
.await;
- assert!(
- result.is_err() || {
- let df = result.unwrap();
- df.collect().await.is_err()
+ let batches = sql_context
+ .sql(&format!(
+ "SELECT id, {projection} FROM paimon.test_db.t ORDER BY id"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let mut rows = Vec::new();
+ for batch in &batches {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let lists = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .unwrap();
+ for row in 0..batch.num_rows() {
+ let values = (!lists.is_null(row)).then(|| {
+ let values = lists.value(row);
+ let blobs =
values.as_any().downcast_ref::<LargeBinaryArray>().unwrap();
+ blobs
+ .iter()
+ .map(|value| value.map(<[u8]>::to_vec))
+ .collect::<Vec<_>>()
+ });
+ rows.push((ids.value(row), values));
+ }
}
- );
+ assert_eq!(
+ rows,
+ vec![
+ (1, Some(vec![Some(b"CC".to_vec()), None])),
+ (2, Some(vec![Some(b"BB".to_vec())])),
+ (3, Some(vec![])),
+ (4, None),
+ ]
+ );
+ if blob_type.starts_with("MAP") {
+ assert_eq!(
+ query_id_name_picture(
+ &sql_context,
+ "SELECT id, 'map' AS name, picture['new'] FROM
paimon.test_db.t WHERE id = 1"
+ )
+ .await,
+ vec![(1, "map".into(), Some(b"CC".to_vec()))]
+ );
+ }
+ }
}
/// Reference: BlobTestBase "Blob: merge-into updates non-blob column on
descriptor blob table"
diff --git a/crates/paimon/src/arrow/format/blob/writer.rs
b/crates/paimon/src/arrow/format/blob/writer.rs
index 5219c5a2..197550f1 100644
--- a/crates/paimon/src/arrow/format/blob/writer.rs
+++ b/crates/paimon/src/arrow/format/blob/writer.rs
@@ -30,14 +30,23 @@ use crate::{Error, Result};
use arrow_array::{
Array, BinaryArray, BooleanArray, Date32Array, Decimal128Array,
Int16Array, Int32Array,
Int64Array, Int8Array, LargeBinaryArray, ListArray, MapArray, RecordBatch,
StringArray,
- Time32MillisecondArray,
+ StructArray, Time32MillisecondArray,
};
use async_trait::async_trait;
use bytes::Bytes;
use crc32fast::Hasher;
use std::collections::{HashMap, HashSet};
+use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
+/// Positions are relative to the logical update range, across rolled files.
+/// Java's BlobPlaceholder has index length -2; an explicit NULL has length -1.
+#[derive(Clone)]
+struct BlobUpdateRows {
+ updated: Arc<[usize]>,
+ next_position: Arc<AtomicUsize>,
+}
+
// Remote FileRead ranges open new requests. Keep read-ahead independent of
// the Java-sized copy buffer so a 4 KiB buffer does not mean 4 KiB GETs.
const SOURCE_READ_SIZE: u64 = 8 * 1024 * 1024;
@@ -56,6 +65,7 @@ pub(crate) struct BlobWriterFactory {
consumer: Option<Arc<dyn BlobConsumer>>,
uri_reader_factory: Option<Arc<dyn UriReaderFactory>>,
copy_buffer_size: usize,
+ update_rows: Option<BlobUpdateRows>,
}
impl BlobWriterFactory {
@@ -72,6 +82,7 @@ impl BlobWriterFactory {
uri_reader_factory: None,
copy_buffer_size: CoreOptions::new(options.unwrap_or(&defaults))
.blob_copy_buffer_size()?,
+ update_rows: None,
})
}
@@ -84,6 +95,16 @@ impl BlobWriterFactory {
self.uri_reader_factory = Some(factory);
self
}
+
+ /// Emit placeholders for unchanged rows of a column with a physical
baseline.
+ pub(crate) fn with_update_rows(mut self, updated: Vec<usize>) -> Self {
+ debug_assert!(updated.windows(2).all(|pair| pair[0] < pair[1]));
+ self.update_rows = Some(BlobUpdateRows {
+ updated: updated.into(),
+ next_position: Arc::new(AtomicUsize::new(0)),
+ });
+ self
+ }
}
#[async_trait]
@@ -107,7 +128,8 @@ impl FormatWriterFactory for BlobWriterFactory {
)?
.with_copy_buffer_size(self.copy_buffer_size)
.with_uri_reader_factory(self.uri_reader_factory.clone())
- .with_consumer(self.consumer.clone()),
+ .with_consumer(self.consumer.clone())
+ .with_update_rows(self.update_rows.clone()),
))
}
}
@@ -124,6 +146,7 @@ pub(crate) struct BlobFormatWriter {
copy_buffer_size: usize,
bytes_written: u64,
lengths: Vec<i64>,
+ update_rows: Option<BlobUpdateRows>,
}
fn invalid(message: impl Into<String>) -> Error {
@@ -167,6 +190,7 @@ impl BlobFormatWriter {
copy_buffer_size: 4 * 1024,
bytes_written: 0,
lengths: Vec::new(),
+ update_rows: None,
})
}
@@ -189,6 +213,11 @@ impl BlobFormatWriter {
self
}
+ fn with_update_rows(mut self, rows: Option<BlobUpdateRows>) -> Self {
+ self.update_rows = rows;
+ self
+ }
+
fn accept(&self, offset: u64, length: i64) -> Result<bool> {
match &self.consumer {
Some(consumer) => consumer.accept(
@@ -466,7 +495,7 @@ impl BlobFormatWriter {
async fn write_collection(
&mut self,
values: &LargeBinaryArray,
- keys: Option<Vec<Vec<u8>>>,
+ keys: Option<Vec<Option<Vec<u8>>>>,
) -> Result<()> {
let (start, mut hasher) = self.start_record().await?;
let (magic, version) = if keys.is_some() {
@@ -484,6 +513,10 @@ impl BlobFormatWriter {
let key_lengths = if let Some(keys) = keys {
let mut lengths = Vec::with_capacity(keys.len());
for key in keys {
+ let Some(key) = key else {
+ lengths.push(-1);
+ continue;
+ };
lengths.push(
i64::try_from(key.len()).map_err(|_| invalid("BLOB MAP key
exceeds i64"))?,
);
@@ -525,7 +558,22 @@ impl FormatFileWriter for BlobFormatWriter {
}
let column = batch.column(0);
let kind = self.kind.clone();
+ let first_position = self
+ .update_rows
+ .as_ref()
+ .map_or(0, |rows| rows.next_position.load(Ordering::Relaxed));
+ let end_position = first_position
+ .checked_add(batch.num_rows())
+ .ok_or_else(|| invalid("BLOB update row position overflows
usize"))?;
for row in 0..batch.num_rows() {
+ if self
+ .update_rows
+ .as_ref()
+ .is_some_and(|rows|
rows.updated.binary_search(&(first_position + row)).is_err())
+ {
+ self.lengths.push(-2);
+ continue;
+ }
if column.is_null(row) {
self.lengths.push(-1);
if let Some(consumer) = &self.consumer {
@@ -548,11 +596,7 @@ impl FormatFileWriter for BlobFormatWriter {
.await?;
}
BlobFieldKind::Map(key_type) => {
- let map = column
- .as_any()
- .downcast_ref::<MapArray>()
- .ok_or_else(|| invalid("MAP<X, BLOB> writer requires
MapArray"))?;
- let entries = map.value(row);
+ let entries = map_entries(column.as_ref(), row)?;
// Java validates all keys before starting the record.
let keys = serialize_map_keys(entries.column(0).as_ref(),
key_type)?;
self.write_collection(blob_values(entries.column(1).as_ref())?, Some(keys))
@@ -560,6 +604,9 @@ impl FormatFileWriter for BlobFormatWriter {
}
}
}
+ if let Some(rows) = &self.update_rows {
+ rows.next_position.store(end_position, Ordering::Relaxed);
+ }
Ok(())
}
@@ -627,12 +674,34 @@ fn validate_key_type(key_type: &DataType) -> Result<()> {
}
}
-fn serialize_map_keys(array: &dyn Array, key_type: &DataType) ->
Result<Vec<Vec<u8>>> {
+/// Arrow Map cannot represent NULL keys. Row updates may transport entries
+/// as a list of key/value structs; physical bytes still use Java's map layout.
+fn map_entries(array: &dyn Array, row: usize) -> Result<StructArray> {
+ if let Some(map) = array.as_any().downcast_ref::<MapArray>() {
+ return Ok(map.value(row));
+ }
+ if let Some(list) = array.as_any().downcast_ref::<ListArray>() {
+ if let Some(entries) =
list.value(row).as_any().downcast_ref::<StructArray>() {
+ if entries.num_columns() == 2 && entries.null_count() == 0 {
+ return Ok(entries.clone());
+ }
+ }
+ }
+ Err(invalid(
+ "MAP<X, BLOB> writer requires MapArray or a list of key/value structs",
+ ))
+}
+
+fn serialize_map_keys(array: &dyn Array, key_type: &DataType) ->
Result<Vec<Option<Vec<u8>>>> {
let mut seen = HashSet::new();
let mut keys = Vec::with_capacity(array.len());
for row in 0..array.len() {
if array.is_null(row) {
- return Err(invalid("Arrow MAP keys must not be null"));
+ if !seen.insert(None) {
+ return Err(invalid("MAP<X, BLOB> keys must be unique"));
+ }
+ keys.push(None);
+ continue;
}
macro_rules! fixed {
($array:ty) => {
@@ -704,10 +773,10 @@ fn serialize_map_keys(array: &dyn Array, key_type:
&DataType) -> Result<Vec<Vec<
})
}
};
- if !seen.insert(key.clone()) {
+ if !seen.insert(Some(key.clone())) {
return Err(invalid("MAP<X, BLOB> keys must be unique"));
}
- keys.push(key);
+ keys.push(Some(key));
}
Ok(keys)
}
@@ -843,7 +912,11 @@ mod tests {
),
];
for (data_type, array) in types {
- let keys = serialize_map_keys(array.as_ref(), &data_type).unwrap();
+ let keys = serialize_map_keys(array.as_ref(), &data_type)
+ .unwrap()
+ .into_iter()
+ .map(Option::unwrap)
+ .collect::<Vec<_>>();
let bytes: Vec<_> =
keys.iter().cloned().map(Bytes::from).collect();
let restored = super::super::decode_blob_map_keys(&bytes,
&data_type).unwrap();
assert_eq!(restored.to_data(), array.to_data(), "{data_type:?}");
@@ -876,4 +949,19 @@ mod tests {
}
assert!(validate_key_type(&DataType::Float(FloatType::new())).is_err());
}
+
+ #[test]
+ fn nullable_map_keys_distinguish_java_null_and_empty_encodings() {
+ let kind = DataType::VarChar(VarCharType::new(10).unwrap());
+ let keys = StringArray::from(vec![None, Some(""), Some("normal")]);
+ assert_eq!(
+ serialize_map_keys(&keys, &kind).unwrap(),
+ vec![None, Some(vec![]), Some(b"normal".to_vec())]
+ );
+ let duplicate = StringArray::from(vec![None::<&str>, None]);
+ assert!(serialize_map_keys(&duplicate, &kind)
+ .unwrap_err()
+ .to_string()
+ .contains("unique"));
+ }
}
diff --git a/crates/paimon/src/arrow/format/blob/writer_consumer_tests.rs
b/crates/paimon/src/arrow/format/blob/writer_consumer_tests.rs
index 010bc5c6..86a46773 100644
--- a/crates/paimon/src/arrow/format/blob/writer_consumer_tests.rs
+++ b/crates/paimon/src/arrow/format/blob/writer_consumer_tests.rs
@@ -57,6 +57,7 @@ impl FileWrite for RecordingOutput {
fn writer(kind: BlobFieldKind, output: &Arc<Mutex<OutputState>>) ->
BlobFormatWriter {
BlobFormatWriter {
+ update_rows: None,
writer: Box::new(RecordingOutput(output.clone())),
file_io: None,
kind,
@@ -148,14 +149,67 @@ async fn
scalar_consumer_runs_after_record_and_null_does_not_flush() {
assert_eq!(writer.lengths, [19, 16, -1]);
}
+#[tokio::test]
+async fn sparse_delta_uses_java_tags_and_only_exposes_updated_values() {
+ let output = Arc::new(Mutex::new(OutputState::default()));
+ let events = Arc::new(Mutex::new(Vec::new()));
+ let mut writer = writer(BlobFieldKind::Scalar, &output)
+ .with_consumer(Some(consumer(&events, &output)))
+ .with_update_rows(Some(BlobUpdateRows {
+ updated: vec![1, 3].into(),
+ next_position: Arc::new(AtomicUsize::new(0)),
+ }));
+ let batch = RecordBatch::try_from_iter([(
+ "payload",
+ Arc::new(LargeBinaryArray::from(vec![
+ Some(b"not copied".as_slice()),
+ None,
+ Some(b"also not copied"),
+ Some(b"new"),
+ ])) as ArrayRef,
+ )])
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ assert_eq!(writer.lengths, [-2, -1, -2, 19]);
+ check_payloads(
+ &output.lock().unwrap(),
+ &events.lock().unwrap(),
+ &[None, Some(b"new")],
+ );
+}
+
+#[tokio::test]
+async fn sparse_cursor_survives_physical_writer_recreation() {
+ let rows = BlobUpdateRows {
+ updated: vec![2].into(),
+ next_position: Arc::new(AtomicUsize::new(0)),
+ };
+ let output = Arc::new(Mutex::new(OutputState::default()));
+ let mut first = writer(BlobFieldKind::Scalar,
&output).with_update_rows(Some(rows.clone()));
+ let batch = RecordBatch::try_from_iter([(
+ "payload",
+ Arc::new(LargeBinaryArray::from(vec![
+ None,
+ None,
+ Some(b"new".as_slice()),
+ ])) as ArrayRef,
+ )])
+ .unwrap();
+ first.write(&batch.slice(0, 2)).await.unwrap();
+ assert_eq!(first.lengths, [-2, -2]);
+ let mut second = writer(BlobFieldKind::Scalar,
&output).with_update_rows(Some(rows));
+ second.write(&batch.slice(2, 1)).await.unwrap();
+ assert_eq!(second.lengths, [19]);
+}
+
#[tokio::test]
async fn collection_consumer_skips_null_elements_and_flushes_after_trailer() {
for keys in [
None,
Some(vec![
- b"key-a".to_vec(),
- b"key-null".to_vec(),
- b"key-b".to_vec(),
+ Some(b"key-a".to_vec()),
+ Some(b"key-null".to_vec()),
+ Some(b"key-b".to_vec()),
]),
] {
let output = Arc::new(Mutex::new(OutputState::default()));
diff --git a/crates/paimon/src/table/data_evolution_blob.rs
b/crates/paimon/src/table/data_evolution_blob.rs
new file mode 100644
index 00000000..ea347aef
--- /dev/null
+++ b/crates/paimon/src/table/data_evolution_blob.rs
@@ -0,0 +1,253 @@
+// 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.
+
+//! Sparse Blob updates compatible with Java BlobFallbackRecordReader.
+
+use super::{matched_column, FileRowRange};
+use crate::spec::{CoreOptions, DataField, DataFileMeta};
+use crate::table::Table;
+use crate::Result;
+use arrow_array::RecordBatch;
+use arrow_select::interleave::interleave;
+use std::collections::HashSet;
+use std::sync::Arc;
+
+pub(super) struct BlobUpdateBatch {
+ pub field: DataField,
+ pub batch: RecordBatch,
+ pub updated_rows: Option<Vec<usize>>,
+}
+
+/// The committer assigns one snapshot sequence to the entire Blob delta.
+pub(super) fn assign_delta_metadata(mut first_row_id: i64, files: &mut
[DataFileMeta]) {
+ for file in files {
+ file.first_row_id = Some(first_row_id);
+ first_row_id += file.row_count;
+ file.min_sequence_number = 0;
+ file.max_sequence_number = 0;
+ }
+}
+
+pub(super) async fn file_updates(
+ table: &Table,
+ fields: &[DataField],
+ batches: &[RecordBatch],
+ matches: &[(usize, usize, usize)],
+ file: &FileRowRange,
+) -> Result<Vec<BlobUpdateBatch>> {
+ let inline =
CoreOptions::new(table.schema().options()).blob_inline_fields();
+ let fields = fields
+ .iter()
+ .filter(|field| is_dedicated_blob(field, &inline))
+ .collect::<Vec<_>>();
+ if fields.is_empty() {
+ return Ok(Vec::new());
+ }
+ let baseline = baseline_field_ids(table, &file.files).await?;
+ fields
+ .into_iter()
+ .map(|field| {
+ let (batch, updated_rows) = delta_batch(
+ field,
+ batches,
+ matches,
+ file.row_count as usize,
+ baseline.contains(&field.id()),
+ )?;
+ Ok(BlobUpdateBatch {
+ field: field.clone(),
+ batch,
+ updated_rows,
+ })
+ })
+ .collect()
+}
+
+pub(super) fn is_dedicated_blob(field: &DataField, inline: &HashSet<String>)
-> bool {
+ field.data_type().is_blob_file_field() && !inline.contains(field.name())
+}
+
+/// The ordinary-column rewrite never requests dedicated Blob payloads. Keep
+/// their independently rolled ranges out of its physical merge plan as well.
+pub(super) fn normal_read_files(files: &[DataFileMeta]) -> Vec<DataFileMeta> {
+ files
+ .iter()
+ .filter(|file| {
+
!crate::table::dedicated_format_file_writer::is_blob_or_video_file_name(&file.file_name)
+ })
+ .cloned()
+ .collect()
+}
+
+/// Resolve providers by field ID, so a dropped column's files cannot supply
+/// a subsequently added Blob column with the same name.
+pub(super) async fn baseline_field_ids(
+ table: &Table,
+ files: &[DataFileMeta],
+) -> Result<HashSet<i32>> {
+ let mut ids = HashSet::new();
+ for file in files {
+ if !file.file_name.to_ascii_lowercase().ends_with(".blob") {
+ continue;
+ }
+ let columns = file
+ .write_cols
+ .as_ref()
+ .filter(|columns| columns.len() == 1)
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!(
+ "Blob file '{}' must identify exactly one write column",
+ file.file_name
+ ),
+ source: None,
+ })?;
+ let stored = if file.schema_id == table.schema().id() {
+ None
+ } else {
+ Some(table.schema_manager().schema(file.schema_id).await?)
+ };
+ let fields = stored
+ .as_ref()
+ .map_or(table.schema().fields(), |schema| schema.fields());
+ ids.extend(
+ super::super::data_evolution_fields::project_by_paths(fields,
columns)?
+ .iter()
+ .map(DataField::id),
+ );
+ }
+ Ok(ids)
+}
+
+/// Only updated payloads enter the Arrow array. Unchanged values are NULL in
+/// memory and receive a distinct placeholder tag when a baseline exists.
+pub(super) fn delta_batch(
+ field: &DataField,
+ batches: &[RecordBatch],
+ matches: &[(usize, usize, usize)],
+ row_count: usize,
+ has_baseline: bool,
+) -> Result<(RecordBatch, Option<Vec<usize>>)> {
+ let length = if has_baseline {
+ matches.last().map_or(0, |(offset, _, _)| offset + 1)
+ } else {
+ row_count
+ };
+ let target = update_arrow_type(field, batches)?;
+ let mut arrays = vec![arrow_array::new_null_array(&target, 1)];
+ for batch in batches {
+ arrays.push(super::super::update_input::cast_update_value(
+ &matched_column(batch, field.name())?,
+ &target,
+ super::super::update_input::CastMode::RowUpdate,
+ )?);
+ }
+ let mut indices = vec![(0, 0); length];
+ for &(offset, batch, row) in matches {
+ if offset >= row_count || offset >= length {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Blob update position {offset} exceeds logical row count
{row_count}"
+ ),
+ source: None,
+ });
+ }
+ if !field.data_type().is_nullable() && arrays[batch + 1].is_null(row) {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Cannot update non-nullable Blob '{}' to
NULL", field.name()),
+ source: None,
+ });
+ }
+ indices[offset] = (batch + 1, row);
+ }
+ if !has_baseline && !field.data_type().is_nullable() && matches.len() !=
row_count {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Non-nullable Blob '{}' has no baseline for unchanged rows",
+ field.name()
+ ),
+ source: None,
+ });
+ }
+ let values = interleave(
+ &arrays
+ .iter()
+ .map(|array| array.as_ref())
+ .collect::<Vec<_>>(),
+ &indices,
+ )
+ .map_err(|error| crate::Error::DataInvalid {
+ message: format!(
+ "Failed to assemble Blob update for '{}': {error}",
+ field.name()
+ ),
+ source: Some(Box::new(error)),
+ })?;
+ // Unchanged positions are NULL in Arrow and placeholders on disk, even
+ // for a non-nullable logical Blob column.
+ let schema =
crate::arrow::build_target_arrow_schema(std::slice::from_ref(field))?;
+ let schema = Arc::new(arrow_schema::Schema::new_with_metadata(
+ vec![schema
+ .field(0)
+ .clone()
+ .with_data_type(target)
+ .with_nullable(true)],
+ schema.metadata().clone(),
+ ));
+ let batch =
+ RecordBatch::try_new(schema, vec![values]).map_err(|error|
crate::Error::DataInvalid {
+ message: format!("Failed to build Blob update batch: {error}"),
+ source: Some(Box::new(error)),
+ })?;
+ Ok((
+ batch,
+ has_baseline.then(|| matches.iter().map(|(offset, _, _)|
*offset).collect()),
+ ))
+}
+
+/// Preserve Java's nullable map keys through a list of key/value structs.
+/// This representation is used only by the row-aware Blob update transport.
+fn update_arrow_type(field: &DataField, batches: &[RecordBatch]) ->
Result<arrow_schema::DataType> {
+ use arrow_schema::DataType;
+ let target = crate::arrow::paimon_type_to_arrow(field.data_type())?;
+ let DataType::Map(entries, _) = &target else {
+ return Ok(target);
+ };
+ let list_input = batches
+ .iter()
+ .map(|batch| matched_column(batch, field.name()))
+ .collect::<Result<Vec<_>>>()?
+ .iter()
+ .any(|array| {
+ matches!(array.data_type(), DataType::List(entry)
+ if matches!(entry.data_type(), DataType::Struct(fields) if
fields.len() == 2))
+ });
+ if !list_input {
+ return Ok(target);
+ }
+ let DataType::Struct(fields) = entries.data_type() else {
+ unreachable!("Arrow map entries are structs");
+ };
+ let fields = vec![
+ Arc::new(fields[0].as_ref().clone().with_nullable(true)),
+ fields[1].clone(),
+ ];
+ Ok(DataType::List(Arc::new(arrow_schema::Field::new(
+ "item",
+ DataType::Struct(fields.into()),
+ false,
+ ))))
+}
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index 74c0c233..6c6a4bcd 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -36,10 +36,10 @@ use crate::table::commit_message::CommitMessage;
use crate::table::data_file_index_writer::FileIndexOptions;
use crate::table::data_file_writer::DataFileWriter;
use crate::table::index_file_path::IndexFileLocation;
-use crate::table::source::data_evolution_anchor_file;
+use crate::table::source::{data_evolution_anchor_file,
is_data_evolution_normal_file};
use crate::table::stats_filter::group_by_overlapping_row_id;
use crate::table::Table;
-use crate::table::{DataSplit, DataSplitBuilder};
+use crate::table::{DataSplit, DataSplitBuilder, RowRange};
use crate::Result;
use arrow_array::{Array, ArrayRef, Int64Array, RecordBatch, StructArray};
use arrow_buffer::NullBuffer;
@@ -54,6 +54,9 @@ use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::Arc;
use uuid::Uuid;
+#[path = "data_evolution_blob.rs"]
+mod blob_updates;
+
const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS";
const DELETION_VECTORS_INDEX_VERSION_V1: u8 = 1;
const MANIFEST_DIR: &str = "manifest";
@@ -81,6 +84,7 @@ pub struct DataEvolutionWriter {
matched_batches: Vec<RecordBatch>,
matched_batch_groups: Vec<usize>,
next_group_id: usize,
+ blob_uri_reader_factory: Option<Arc<dyn crate::io::UriReaderFactory>>,
}
impl DataEvolutionWriter {
@@ -148,9 +152,7 @@ impl DataEvolutionWriter {
});
}
let partition_keys = schema.partition_keys();
- // Java CoreOptions.updatableBlobFields() includes descriptor and view
- // fields, both of which store their references in normal data files.
- let updatable_blob_fields = core_options.blob_inline_fields();
+ let video_fields = core_options.video_frame_fields();
for col in &update_columns {
let top_level =
DataEvolutionPartialWriter::top_level_write_name(col, schema.fields());
if !allow_partition_columns && partition_keys.iter().any(|key| key
== top_level) {
@@ -158,17 +160,10 @@ impl DataEvolutionWriter {
message: format!("Cannot update partition column '{col}'
in MERGE INTO"),
});
}
- if let Some(field) = schema.fields().iter().find(|f| f.name() ==
top_level) {
- if field.data_type().is_blob_file_field()
- && !updatable_blob_fields.contains(top_level)
- {
- return Err(crate::Error::Unsupported {
- message: format!(
- "Cannot update raw-data BLOB column '{col}' in
MERGE INTO. \
- Only inline BLOB columns listed in
'blob-descriptor-field' or 'blob-view-field' can be updated"
- ),
- });
- }
+ if video_fields.contains(top_level) {
+ return Err(crate::Error::Unsupported {
+ message: format!("Updating packed VIDEO column '{col}' is
not supported"),
+ });
}
}
@@ -180,9 +175,17 @@ impl DataEvolutionWriter {
matched_batches: Vec::new(),
matched_batch_groups: Vec::new(),
next_group_id: 1,
+ blob_uri_reader_factory: None,
})
}
+ pub(super) fn with_blob_uri_reader_factory(
+ &mut self,
+ factory: Option<Arc<dyn crate::io::UriReaderFactory>>,
+ ) {
+ self.blob_uri_reader_factory = factory;
+ }
+
fn validate_partition_values(
&self,
files: &[FileRowRange],
@@ -351,6 +354,14 @@ impl DataEvolutionWriter {
// 3. For each affected file: read original columns, apply updates,
write partial files
let mut writer = DataEvolutionPartialWriter::new(&self.table,
self.update_columns.clone())?;
+ writer.blob_uri_reader_factory = self.blob_uri_reader_factory.clone();
+ let inline =
CoreOptions::new(self.table.schema().options()).blob_inline_fields();
+ let normal_fields = self
+ .write_fields
+ .iter()
+ .filter(|field| !blob_updates::is_dedicated_blob(field, &inline))
+ .cloned()
+ .collect::<Vec<_>>();
let result = async {
for (&file_pos, matched_rows) in &file_matches {
@@ -358,22 +369,63 @@ impl DataEvolutionWriter {
let first_row_id = file_range.first_row_id;
let row_count = file_range.row_count as usize;
+ let mut sorted_matches: Vec<(usize, usize, usize)> =
matched_rows
+ .iter()
+ .map(|m| (m.offset, m.batch_idx, m.row_idx))
+ .collect();
+ sorted_matches.sort_by_key(|(offset, _, _)| *offset);
+ if sorted_matches.windows(2).any(|rows| rows[0].0 ==
rows[1].0) {
+ return Err(crate::Error::DataInvalid {
+ message: "Multiple updates target the same row
ID".to_string(),
+ source: None,
+ });
+ }
+ let blobs = blob_updates::file_updates(
+ &self.table,
+ &self.write_fields,
+ &self.matched_batches,
+ &sorted_matches,
+ file_range,
+ )
+ .await?;
+ if normal_fields.is_empty() {
+ for blob in blobs {
+ writer
+ .write_blob_batch(
+ file_range,
+ &blob.field,
+ blob.batch,
+ blob.updated_rows,
+ )
+ .await?;
+ }
+ continue;
+ }
+
// Read original columns from the entire file group (base +
partial-column files).
- let col_refs: Vec<&str> =
self.write_fields.iter().map(DataField::name).collect();
+ let mut col_refs: Vec<&str> =
normal_fields.iter().map(DataField::name).collect();
+ // Anchor row counts even when all ordinary SET columns were
+ // added after the original files were written.
+ col_refs.push("_ROW_ID");
let mut rb = read_table.new_read_builder();
rb.with_projection(&col_refs)?;
+ rb.with_row_ranges(vec![RowRange::new(
+ file_range.first_row_id,
+ file_range.last_row_id,
+ )]);
let read = rb.new_read()?;
// Base + partial-column files share row-id ranges, so physical
// row counts overcount the group's logical rows.
+ let read_files =
blob_updates::normal_read_files(&file_range.files);
let split = DataSplitBuilder::new()
.with_snapshot(file_range.snapshot_id)
.with_partition(BinaryRow::from_serialized_bytes(&file_range.partition)?)
.with_bucket(file_range.bucket)
.with_bucket_path(file_range.bucket_path.clone())
.with_total_buckets(file_range.total_buckets)
- .with_data_files(file_range.files.clone())
- .with_raw_convertible(file_range.files.len() == 1)
+ .with_raw_convertible(read_files.len() == 1)
+ .with_data_files(read_files)
.build()?;
let stream = read.to_arrow(&[split])?;
@@ -405,19 +457,7 @@ impl DataEvolutionWriter {
// Keep one physical ROW column per top-level field, even when
the
// update names identify several nested leaves of that ROW.
- let mut sorted_matches: Vec<(usize, usize, usize)> =
matched_rows
- .iter()
- .map(|m| (m.offset, m.batch_idx, m.row_idx))
- .collect();
- sorted_matches.sort_by_key(|(offset, _, _)| *offset);
- if sorted_matches.windows(2).any(|rows| rows[0].0 ==
rows[1].0) {
- return Err(crate::Error::DataInvalid {
- message: "Multiple updates target the same row
ID".to_string(),
- source: None,
- });
- }
- let new_columns = self
- .write_fields
+ let new_columns = normal_fields
.iter()
.enumerate()
.map(|(index, field)| {
@@ -431,7 +471,7 @@ impl DataEvolutionWriter {
)
})
.collect::<Result<Vec<_>>>()?;
- let write_schema =
crate::arrow::build_target_arrow_schema(&self.write_fields)?;
+ let write_schema =
crate::arrow::build_target_arrow_schema(&normal_fields)?;
let updated_batch =
RecordBatch::try_new(write_schema,
new_columns).map_err(|e| {
crate::Error::DataInvalid {
@@ -449,6 +489,11 @@ impl DataEvolutionWriter {
updated_batch,
)
.await?;
+ for blob in blobs {
+ writer
+ .write_blob_batch(file_range, &blob.field, blob.batch,
blob.updated_rows)
+ .await?;
+ }
}
// 4. Collect commit messages (caller is responsible for
committing)
@@ -585,7 +630,7 @@ impl DataEvolutionDeleteWriter {
.cloned()
.collect::<Vec<_>>();
- for group in group_by_overlapping_row_id(files) {
+ for group in group_by_normal_row_id(files) {
let anchor = data_evolution_anchor_file(&group)?;
let first_row_id =
anchor
@@ -1232,18 +1277,23 @@ impl RowIdFileIndex {
.cloned()
.collect();
- let groups = group_by_overlapping_row_id(all_files);
+ let groups = group_by_normal_row_id(all_files);
for group in groups {
// Compute the overall row_id range for this group.
// The base file has the widest range; partial-column files
share it.
- let first_row_id = group.iter().filter_map(|f|
f.first_row_id).min().unwrap();
+ let normal = group
+ .iter()
+ .filter(|file| is_data_evolution_normal_file(file))
+ .collect::<Vec<_>>();
+ let first_row_id = normal.iter().filter_map(|f|
f.first_row_id).min().unwrap();
let last_row_id = group
.iter()
+ .filter(|file| is_data_evolution_normal_file(file))
.filter_map(|f| f.row_id_range().map(|(_, end)| end))
.max()
.unwrap();
// The actual row count is the max among the group (base
file's count).
- let row_count = group.iter().map(|f|
f.row_count).max().unwrap();
+ let row_count = normal.iter().map(|f|
f.row_count).max().unwrap();
file_index.push(FileRowRange {
first_row_id,
@@ -1284,6 +1334,38 @@ impl RowIdFileIndex {
}
}
+/// Dedicated column files can cross normal-file boundaries. They provide
+/// values, but must not merge anchors or introduce otherwise absent row IDs.
+fn group_by_normal_row_id(files: Vec<DataFileMeta>) -> Vec<Vec<DataFileMeta>> {
+ let (normal, dedicated): (Vec<_>, Vec<_>) =
+ files.into_iter().partition(is_data_evolution_normal_file);
+ group_by_overlapping_row_id(normal)
+ .into_iter()
+ .map(|mut group| {
+ let first = group
+ .iter()
+ .filter_map(|file| file.first_row_id)
+ .min()
+ .unwrap();
+ let last = group
+ .iter()
+ .filter_map(|file| file.row_id_range().map(|(_, last)| last))
+ .max()
+ .unwrap();
+ group.extend(
+ dedicated
+ .iter()
+ .filter(|file| {
+ file.row_id_range()
+ .is_some_and(|(from, to)| from <= last && to >=
first)
+ })
+ .cloned(),
+ );
+ group
+ })
+ .collect()
+}
+
struct FileRowRange {
first_row_id: i64,
last_row_id: i64,
@@ -1304,7 +1386,7 @@ struct MatchedRow {
}
// ---------------------------------------------------------------------------
-// DataEvolutionPartialWriter — writes partial-column parquet files for data
evolution
+// DataEvolutionPartialWriter — writes partial-column data evolution files
// ---------------------------------------------------------------------------
/// Key: (partition_bytes, bucket, first_row_id)
@@ -1324,6 +1406,8 @@ type PartialCommitGroup = (Option<i64>,
Vec<DataFileMeta>);
/// `file_source = APPEND (0)`, caller-supplied `first_row_id`, and
`write_cols`.
/// VECTOR columns use dedicated `*.vector.<format>` files when
`vector.file.format`
/// is configured.
+/// Raw BLOB, ARRAY<BLOB> and MAP<X, BLOB> columns use dedicated `*.blob`
+/// files with Java's placeholder tags for unchanged rows.
pub(crate) struct DataEvolutionPartialWriter {
file_io: FileIO,
table_location: String,
@@ -1338,12 +1422,14 @@ pub(crate) struct DataEvolutionPartialWriter {
/// Writers keyed by (partition_bytes, bucket, first_row_id, file kind).
writers: HashMap<WriterKey, DataFileWriter>,
check_from_snapshots: HashMap<WriterBaseKey, i64>,
+ blob_uri_reader_factory: Option<Arc<dyn crate::io::UriReaderFactory>>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
enum PartialFileKind {
Normal,
Vector,
+ Blob(i32),
}
#[derive(Clone)]
@@ -1354,7 +1440,6 @@ struct PartialWriteSet {
write_fields: Vec<DataField>,
write_columns: Vec<String>,
omit_write_cols: bool,
- column_indices: Vec<usize>,
schema: Arc<arrow_schema::Schema>,
file_index_options: Option<Arc<FileIndexOptions>>,
}
@@ -1383,10 +1468,16 @@ impl DataEvolutionPartialWriter {
.as_ref()
.and_then(|options|
options.project_to_fields(&write_set.write_fields))
.map(Arc::new);
- if write_set.kind == PartialFileKind::Vector &&
write_set.file_index_options.is_some() {
+ if write_set.kind != PartialFileKind::Normal &&
write_set.file_index_options.is_some() {
+ let kind = if write_set.kind == PartialFileKind::Vector {
+ "VECTOR"
+ } else {
+ "BLOB"
+ };
return Err(crate::Error::Unsupported {
- message: "FileIndex generation does not support dedicated
VECTOR partial files"
- .to_string(),
+ message: format!(
+ "FileIndex generation does not support dedicated
{kind} partial files"
+ ),
});
}
}
@@ -1410,6 +1501,7 @@ impl DataEvolutionPartialWriter {
write_sets,
writers: HashMap::new(),
check_from_snapshots: HashMap::new(),
+ blob_uri_reader_factory: None,
})
}
@@ -1437,17 +1529,19 @@ impl DataEvolutionPartialWriter {
)?;
let projected = super::data_evolution_fields::project_by_paths(fields,
write_columns)?;
let mut normal_fields = Vec::new();
- let mut normal_indices = Vec::new();
let mut vector_fields = Vec::new();
- let mut vector_indices = Vec::new();
-
- for (idx, field) in projected.into_iter().enumerate() {
- if vector_file_format.is_some() && matches!(field.data_type(),
DataType::Vector(_)) {
+ let inline = core_options.blob_inline_fields();
+ let mut blob_fields = Vec::new();
+
+ for field in projected {
+ if blob_updates::is_dedicated_blob(&field, &inline) {
+ blob_fields.push(field);
+ } else if vector_file_format.is_some()
+ && matches!(field.data_type(), DataType::Vector(_))
+ {
vector_fields.push(field);
- vector_indices.push(idx);
} else {
normal_fields.push(field);
- normal_indices.push(idx);
}
}
@@ -1484,7 +1578,6 @@ impl DataEvolutionPartialWriter {
core_options,
),
write_columns: normal_columns,
- column_indices: normal_indices,
schema,
file_index_options: None,
});
@@ -1500,12 +1593,24 @@ impl DataEvolutionPartialWriter {
write_fields: vector_fields,
write_columns: vector_columns,
omit_write_cols: false,
- column_indices: vector_indices,
schema,
file_index_options: None,
});
}
+ for field in blob_fields {
+ let schema =
crate::arrow::build_target_arrow_schema(std::slice::from_ref(&field))?;
+ write_sets.push(PartialWriteSet {
+ kind: PartialFileKind::Blob(field.id()),
+ target_file_size: core_options.blob_target_file_size(),
+ file_format: "blob".to_string(),
+ write_columns: vec![field.name().to_string()],
+ write_fields: vec![field],
+ omit_write_cols: false,
+ schema,
+ file_index_options: None,
+ });
+ }
Ok(write_sets)
}
@@ -1547,6 +1652,9 @@ impl DataEvolutionPartialWriter {
};
for write_set in self.write_sets.clone() {
+ if matches!(write_set.kind, PartialFileKind::Blob(_)) {
+ continue;
+ }
let key = (
partition_bytes.clone(),
bucket,
@@ -1554,33 +1662,8 @@ impl DataEvolutionPartialWriter {
write_set.kind,
);
if !self.writers.contains_key(&key) {
- let writer = DataFileWriter::new(
- self.file_io.clone(),
- self.table_location.clone(),
- partition_path.clone(),
- bucket,
- self.schema_id,
- write_set.target_file_size,
- self.file_compression.clone(),
- self.file_compression_zstd_level,
- self.write_buffer_size,
- write_set.file_format.clone(),
- write_set.write_fields.clone(),
- self.format_options.clone(),
- Some(0), // file_source: APPEND
- Some(first_row_id),
- (!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
- };
+ let writer =
+ self.new_file_writer(&write_set, partition_path.clone(),
bucket, first_row_id)?;
self.writers.insert(key.clone(), writer);
}
@@ -1592,12 +1675,104 @@ impl DataEvolutionPartialWriter {
Ok(())
}
+ async fn write_blob_batch(
+ &mut self,
+ file: &FileRowRange,
+ field: &DataField,
+ batch: RecordBatch,
+ updated_rows: Option<Vec<usize>>,
+ ) -> Result<()> {
+ let write_set = self
+ .write_sets
+ .iter()
+ .find(|set| set.kind == PartialFileKind::Blob(field.id()))
+ .expect("dedicated Blob field has a physical write set")
+ .clone();
+ let base_key = (file.partition.clone(), file.bucket,
file.first_row_id);
+ self.check_from_snapshots
+ .entry(base_key)
+ .or_insert(file.snapshot_id);
+ let key = (
+ file.partition.clone(),
+ file.bucket,
+ file.first_row_id,
+ write_set.kind,
+ );
+ if !self.writers.contains_key(&key) {
+ let partition_path = if self.partition_keys.is_empty() {
+ String::new()
+ } else {
+ self.partition_computer
+
.generate_partition_path(&BinaryRow::from_serialized_bytes(&file.partition)?)?
+ };
+ let mut writer =
+ self.new_file_writer(&write_set, partition_path, file.bucket,
file.first_row_id)?;
+ if let Some(rows) = updated_rows {
+ writer.set_blob_update_rows(rows,
self.blob_uri_reader_factory.clone())?;
+ } else if self.blob_uri_reader_factory.is_some() {
+ writer.set_blob_writer_options(None,
self.blob_uri_reader_factory.clone())?;
+ }
+ self.writers.insert(key.clone(), writer);
+ }
+ let writer = self.writers.get_mut(&key).unwrap();
+ // Dedicated Blob writers check payload bytes after each record.
+ for row in 0..batch.num_rows() {
+ writer.write(&batch.slice(row, 1)).await?;
+ }
+ Ok(())
+ }
+
+ fn new_file_writer(
+ &self,
+ write_set: &PartialWriteSet,
+ partition_path: String,
+ bucket: i32,
+ first_row_id: i64,
+ ) -> Result<DataFileWriter> {
+ let writer = DataFileWriter::new(
+ self.file_io.clone(),
+ self.table_location.clone(),
+ partition_path,
+ bucket,
+ self.schema_id,
+ write_set.target_file_size,
+ self.file_compression.clone(),
+ self.file_compression_zstd_level,
+ self.write_buffer_size,
+ write_set.file_format.clone(),
+ write_set.write_fields.clone(),
+ self.format_options.clone(),
+ Some(0),
+ Some(first_row_id),
+ (!write_set.omit_write_cols).then(||
write_set.write_columns.clone()),
+ )?
+ .with_file_index(write_set.file_index_options.clone());
+ Ok(
+ if self
+ .write_sets
+ .iter()
+ .any(|set| set.kind == PartialFileKind::Vector)
+ {
+ writer.without_row_sidecar()
+ } else {
+ writer
+ },
+ )
+ }
+
fn project_batch(batch: &RecordBatch, write_set: &PartialWriteSet) ->
Result<RecordBatch> {
let columns: Vec<ArrayRef> = write_set
- .column_indices
+ .write_fields
.iter()
- .map(|&idx| batch.column(idx).clone())
- .collect();
+ .map(|field| {
+ batch.column_by_name(field.name()).cloned().ok_or_else(|| {
+ crate::Error::DataInvalid {
+ message: format!("Missing data-evolution write column
'{}'", field.name()),
+ source: None,
+ }
+ })
+ })
+ .collect::<Result<Vec<_>>>()?;
RecordBatch::try_new(write_set.schema.clone(), columns).map_err(|e| {
crate::Error::DataInvalid {
message: format!(
@@ -1626,7 +1801,10 @@ impl DataEvolutionPartialWriter {
// Group files by (partition, bucket) since multiple first_row_ids may
share the same partition/bucket
let mut grouped: HashMap<(Vec<u8>, i32), PartialCommitGroup> =
HashMap::new();
- for ((partition_bytes, bucket, first_row_id, _kind), files) in results
{
+ for ((partition_bytes, bucket, first_row_id, kind), mut files) in
results {
+ if matches!(kind, PartialFileKind::Blob(_)) {
+ blob_updates::assign_delta_metadata(first_row_id, &mut files);
+ }
let base_key = (partition_bytes.clone(), bucket, first_row_id);
let check_from_snapshot =
check_from_snapshots.get(&base_key).copied();
let entry = grouped
@@ -2721,6 +2899,25 @@ mod tests {
assert_eq!(row_count, 100);
}
+ #[test]
+ fn dedicated_ranges_attach_to_normal_anchors_without_bridging_gaps() {
+ let groups = group_by_normal_row_id(vec![
+ make_test_file_meta("first.parquet", 2, Some(0), 1, None),
+ make_test_file_meta("second.parquet", 2, Some(4), 1, None),
+ make_test_file_meta("payload.blob", 8, Some(0), 1,
Some(vec!["payload".into()])),
+ make_test_file_meta("other.blob", 1, Some(10), 1,
Some(vec!["payload".into()])),
+ ]);
+ assert_eq!(groups.len(), 2);
+ for (group, expected) in groups.iter().zip([(0, 1), (4, 5)]) {
+ assert_eq!(group.len(), 2);
+ assert_eq!(
+ data_evolution_anchor_file(group).unwrap().row_id_range(),
+ Some(expected)
+ );
+ assert_eq!(group[1].file_name, "payload.blob");
+ }
+ }
+
#[tokio::test]
async fn test_rejects_non_data_evolution_table() {
let file_io = test_file_io();
diff --git a/crates/paimon/src/table/data_file_writer.rs
b/crates/paimon/src/table/data_file_writer.rs
index 14efe59d..099b9dea 100644
--- a/crates/paimon/src/table/data_file_writer.rs
+++ b/crates/paimon/src/table/data_file_writer.rs
@@ -170,11 +170,7 @@ impl DataFileWriter {
uri_reader_factory: Option<Arc<dyn crate::io::UriReaderFactory>>,
) -> Result<()> {
debug_assert!(self.current_writer.is_none());
- let mut factory = crate::arrow::format::blob::BlobWriterFactory::new(
- Some(self.file_io.clone()),
- self.write_fields.first(),
- Some(&self.format_options),
- )?;
+ let mut factory = self.blob_writer_factory()?;
// Java BlobFormatWriter relinquishes deletion rights whenever a
// consumer can expose a descriptor, even if a later callback fails.
self.delete_file_upon_abort = consumer.is_none();
@@ -188,6 +184,28 @@ impl DataFileWriter {
Ok(())
}
+ fn blob_writer_factory(&self) ->
Result<crate::arrow::format::blob::BlobWriterFactory> {
+ crate::arrow::format::blob::BlobWriterFactory::new(
+ Some(self.file_io.clone()),
+ self.write_fields.first(),
+ Some(&self.format_options),
+ )
+ }
+
+ pub(super) fn set_blob_update_rows(
+ &mut self,
+ updated: Vec<usize>,
+ reader_factory: Option<Arc<dyn crate::io::UriReaderFactory>>,
+ ) -> Result<()> {
+ debug_assert!(self.current_writer.is_none());
+ let mut factory =
self.blob_writer_factory()?.with_update_rows(updated);
+ if let Some(reader_factory) = reader_factory {
+ factory = factory.with_uri_reader_factory(reader_factory);
+ }
+ self.format_writer_factory = Some(Arc::new(factory));
+ Ok(())
+ }
+
/// 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 {
diff --git a/crates/paimon/src/table/table_update_by_row_id.rs
b/crates/paimon/src/table/table_update_by_row_id.rs
index c320ca60..afec5f00 100644
--- a/crates/paimon/src/table/table_update_by_row_id.rs
+++ b/crates/paimon/src/table/table_update_by_row_id.rs
@@ -39,6 +39,7 @@ pub struct TableUpdateByRowId {
index: RowIdFileIndex,
updated: HashMap<i32, HashSet<i64>>,
messages: Vec<CommitMessage>,
+ blob_uri_reader_factory: Option<std::sync::Arc<dyn
crate::io::UriReaderFactory>>,
}
impl TableUpdateByRowId {
@@ -54,9 +55,20 @@ impl TableUpdateByRowId {
index,
updated: HashMap::new(),
messages: Vec::new(),
+ blob_uri_reader_factory: None,
})
}
+ /// URI reader selection for referenced Blob values. Core performs the
+ /// bounded or streamed copy; callers need not materialize payloads.
+ pub fn with_blob_uri_reader_factory(
+ &mut self,
+ factory: Option<std::sync::Arc<dyn crate::io::UriReaderFactory>>,
+ ) -> &mut Self {
+ self.blob_uri_reader_factory = factory;
+ self
+ }
+
/// Stage one logical Arrow table. Its chunks may share a file group.
/// Partition columns may carry their existing values. Changing them is
/// rejected before writing, because a move requires delete + insert.
@@ -74,6 +86,7 @@ impl TableUpdateByRowId {
.filter(|name| seen.insert(name.clone()))
.collect::<Vec<_>>();
let mut writer = DataEvolutionWriter::for_row_id(&self.table,
columns.clone())?;
+
writer.with_blob_uri_reader_factory(self.blob_uri_reader_factory.clone());
let batches = batches
.into_iter()
.map(super::update_input::normalize_row_ids)
diff --git a/crates/paimon/tests/blob_collection_write_test.rs
b/crates/paimon/tests/blob_collection_write_test.rs
index 0455d9a9..fa5df004 100644
--- a/crates/paimon/tests/blob_collection_write_test.rs
+++ b/crates/paimon/tests/blob_collection_write_test.rs
@@ -209,20 +209,6 @@ async fn
collections_use_java_blob_files_and_preserve_null_empty_and_sliced_valu
}
}
-#[tokio::test]
-async fn raw_blob_collection_updates_are_rejected_like_java() {
- let table = collection_table(&[]).await;
- for column in ["items", "attrs"] {
- assert!(
- table
- .new_write_builder()
- .new_data_evolution_writer(vec![column.into()])
- .is_err(),
- "raw Blob collection {column} must not be updated"
- );
- }
-}
-
#[test]
fn video_configuration_matches_java_and_does_not_fall_through_to_blob_writes()
{
fn schema(
diff --git a/crates/paimon/tests/blob_update_test.rs
b/crates/paimon/tests/blob_update_test.rs
new file mode 100644
index 00000000..a9f5346e
--- /dev/null
+++ b/crates/paimon/tests/blob_update_test.rs
@@ -0,0 +1,890 @@
+// 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::builder::{LargeBinaryBuilder, ListBuilder, MapBuilder,
StringBuilder};
+use arrow_array::{
+ Array, ArrayRef, Int32Array, Int64Array, LargeBinaryArray, ListArray,
MapArray, RecordBatch,
+};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+use futures::TryStreamExt;
+use paimon::spec::{
+ ArrayType, BlobDescriptor, BlobType, DataType, IntType, MapType, Schema,
TableSchema,
+ VarCharType,
+};
+use paimon::table::Table;
+use std::collections::HashSet;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+
+async fn table(options: &[(&str, &str)]) -> Table {
+ typed_table(DataType::Blob(BlobType::new()), options).await
+}
+
+async fn typed_table(payload_type: DataType, options: &[(&str, &str)]) ->
Table {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", payload_type)
+ .column("value", DataType::Int(IntType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option("deletion-vectors.enabled", "true");
+ for (key, value) in options {
+ schema = schema.option(*key, *value);
+ }
+ let path = "memory:/blob_updates";
+ let (io, table) = memory_table(path, TableSchema::new(0,
&schema.build().unwrap()));
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ table
+}
+
+fn matched(ids: Vec<i64>, values: Vec<Option<&[u8]>>) -> RecordBatch {
+ RecordBatch::try_from_iter([
+ ("_ROW_ID", Arc::new(Int64Array::from(ids)) as ArrayRef),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from(values)) as ArrayRef,
+ ),
+ ])
+ .unwrap()
+}
+
+async fn apply(table: &Table, batch: RecordBatch) ->
Vec<paimon::table::CommitMessage> {
+ table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![batch])
+ .await
+ .unwrap()
+}
+
+async fn read_batch(table: &Table) -> RecordBatch {
+ let builder = table.new_read_builder();
+ 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();
+ arrow_select::concat::concat_batches(&batches[0].schema(),
&batches).unwrap()
+}
+
+#[tokio::test]
+async fn
short_deltas_keep_prefix_placeholders_and_stop_before_unchanged_tail() {
+ let table = table(&[]).await;
+ seed(&table).await;
+ let builder = table.new_write_builder();
+ let messages = apply(&table, matched(vec![1], vec![Some(b"first")])).await;
+ assert_eq!(messages[0].new_files[0].row_count, 2);
+ builder.new_commit().commit(messages).await.unwrap();
+ let messages = apply(&table, matched(vec![2],
vec![Some(b"second")])).await;
+ assert_eq!(messages[0].new_files[0].row_count, 3);
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(
+ rows(&table)
+ .await
+ .iter()
+ .map(|row| row.1.clone())
+ .collect::<Vec<_>>(),
+ vec![
+ Some(b"old-0".to_vec()),
+ Some(b"first".to_vec()),
+ Some(b"second".to_vec()),
+ Some(b"old-3".to_vec()),
+ Some(b"old-4".to_vec()),
+ ]
+ );
+ let historic = table.copy_with_options(
+ [("scan.snapshot-id".into(), "1".into())]
+ .into_iter()
+ .collect(),
+ );
+ assert_eq!(rows(&historic).await[1].1, None);
+ let mut read = table.new_read_builder();
+ read.with_projection(&["payload"])
+ .unwrap()
+ .with_row_ranges(vec![paimon::table::RowRange::new(1, 2)]);
+ let plan = read.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let batch = arrow_select::concat::concat_batches(&batches[0].schema(),
&batches).unwrap();
+ let payloads = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap();
+ assert_eq!(
+ payloads.iter().collect::<Vec<_>>(),
+ vec![Some(b"first".as_slice()), Some(b"second".as_slice())]
+ );
+}
+
+#[tokio::test]
+async fn sparse_positions_continue_across_rolled_delta_files() {
+ let table = table(&[("blob.target-file-size", "20 B")]).await;
+ seed(&table).await;
+ let messages = apply(
+ &table,
+ matched(vec![0, 4], vec![Some(b"first"), Some(b"last")]),
+ )
+ .await;
+ let mut ranges = messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .map(|file| file.row_id_range().unwrap())
+ .collect::<Vec<_>>();
+ ranges.sort_unstable();
+ assert_eq!(ranges, vec![(0, 0), (1, 4)]);
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(
+ rows(&table)
+ .await
+ .iter()
+ .map(|row| row.1.clone())
+ .collect::<Vec<_>>(),
+ vec![
+ Some(b"first".to_vec()),
+ None,
+ Some(vec![]),
+ Some(b"old-3".to_vec()),
+ Some(b"last".to_vec()),
+ ]
+ );
+}
+
+#[tokio::test]
+async fn absent_blob_baseline_writes_real_nulls_for_the_entire_normal_range() {
+ let table = table(&[]).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .with_write_type(vec!["id".into(), "value".into()])
+ .unwrap();
+ writer
+ .write_arrow_batch(
+ &RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![0, 1, 2])) as ArrayRef),
+ (
+ "value",
+ Arc::new(Int32Array::from(vec![10, 11, 12])) as ArrayRef,
+ ),
+ ])
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ builder
+ .new_commit()
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ let messages = apply(&table, matched(vec![1], vec![Some(b"new")])).await;
+ assert_eq!(
+ (
+ messages[0].new_files[0].first_row_id,
+ messages[0].new_files[0].row_count
+ ),
+ (Some(0), 3)
+ );
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(
+ rows(&table).await,
+ vec![(0, None, 10), (1, Some(b"new".to_vec()), 11), (2, None, 12)]
+ );
+}
+
+#[tokio::test]
+async fn descriptor_updates_copy_only_matched_payloads() {
+ let table = table(&[]).await;
+ seed(&table).await;
+ let source = "memory:/external";
+ table
+ .file_io()
+ .new_output(source)
+ .unwrap()
+ .write(b"ignoredreplacementignored".to_vec().into())
+ .await
+ .unwrap();
+ let descriptor = BlobDescriptor::new(source.into(), 7, 11).serialize();
+ let messages = apply(&table, matched(vec![3],
vec![Some(&descriptor)])).await;
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+ table.file_io().delete_file(source).await.unwrap();
+ assert_eq!(rows(&table).await[3].1, Some(b"replacement".to_vec()));
+}
+
+#[tokio::test]
+async fn non_nullable_blob_allows_placeholders_but_rejects_explicit_null() {
+ let table = typed_table(
+ DataType::Blob(BlobType::new())
+ .copy_with_nullable(false)
+ .unwrap(),
+ &[],
+ )
+ .await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(
+ &RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![0, 1, 2])) as ArrayRef),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from(vec![
+ b"zero".as_slice(),
+ b"one",
+ b"two",
+ ])) as ArrayRef,
+ ),
+ (
+ "value",
+ Arc::new(Int32Array::from(vec![10, 11, 12])) as ArrayRef,
+ ),
+ ])
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ builder
+ .new_commit()
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ let messages = apply(&table, matched(vec![1], vec![Some(b"new")])).await;
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(rows(&table).await[0].1, Some(b"zero".to_vec()));
+ let error = builder
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![matched(vec![1], vec![None])])
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("NULL"), "{error}");
+ assert_eq!(rows(&table).await[1].1, Some(b"new".to_vec()));
+}
+
+fn collection_values(map: bool, values: &[Option<Vec<Option<&[u8]>>>]) ->
ArrayRef {
+ if map {
+ let mut builder = MapBuilder::new(None, StringBuilder::new(),
LargeBinaryBuilder::new());
+ for row in values {
+ if let Some(row) = row {
+ for (index, value) in row.iter().enumerate() {
+ builder.keys().append_value(index.to_string());
+ builder.values().append_option(*value);
+ }
+ }
+ builder.append(row.is_some()).unwrap();
+ }
+ Arc::new(builder.finish())
+ } else {
+ let mut builder = ListBuilder::new(LargeBinaryBuilder::new());
+ for row in values {
+ if let Some(row) = row {
+ for value in row {
+ builder.values().append_option(*value);
+ }
+ }
+ builder.append(row.is_some());
+ }
+ Arc::new(builder.finish())
+ }
+}
+
+#[tokio::test]
+async fn collection_updates_preserve_parent_null_empty_and_child_null_values()
{
+ for map in [false, true] {
+ for rolling in [false, true] {
+ let blob = DataType::Blob(BlobType::new());
+ let kind = if map {
+ DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::new(100).unwrap()),
+ blob,
+ ))
+ } else {
+ DataType::Array(ArrayType::new(blob))
+ };
+ let table = typed_table(
+ kind,
+ &[(
+ "blob.target-file-size",
+ if rolling { "20 B" } else { "1 MB" },
+ )],
+ )
+ .await;
+ let original = collection_values(
+ map,
+ &[
+ Some(vec![Some(b"zero"), None, Some(b"")]),
+ None,
+ Some(vec![]),
+ Some(vec![Some(b"three")]),
+ Some(vec![Some(b"four")]),
+ ],
+ );
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(
+ &RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![0, 1, 2, 3, 4])) as
ArrayRef,
+ ),
+ ("payload", original.clone()),
+ (
+ "value",
+ Arc::new(Int32Array::from(vec![10, 11, 12, 13,
14])) as ArrayRef,
+ ),
+ ])
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ builder
+ .new_commit()
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ // Sliced input exercises non-zero list/map offsets as well as
+ // unordered row IDs spanning separate input batches.
+ let values = collection_values(
+ map,
+ &[
+ Some(vec![Some(b"unused")]),
+ Some(vec![Some(b"new"), None, Some(b"")]),
+ None,
+ ],
+ );
+ let batch = RecordBatch::try_from_iter([
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from(vec![-1, 4, 1])) as ArrayRef,
+ ),
+ ("payload", values),
+ ])
+ .unwrap()
+ .slice(1, 2);
+ let messages = apply(&table, batch).await;
+ assert!(messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .all(|file| file.file_name.ends_with(".blob")));
+ builder.new_commit().commit(messages).await.unwrap();
+ let batch = read_batch(&table).await;
+ let expected = collection_values(
+ map,
+ &[
+ Some(vec![Some(b"zero"), None, Some(b"")]),
+ None,
+ Some(vec![]),
+ Some(vec![Some(b"three")]),
+ Some(vec![Some(b"new"), None, Some(b"")]),
+ ],
+ );
+ let expected = arrow_cast::cast(&expected,
batch.column(1).data_type()).unwrap();
+ assert_eq!(batch.column(1).to_data(), expected.to_data());
+ let descriptors = table.copy_with_options(
+ [("blob-as-descriptor".into(), "true".into())]
+ .into_iter()
+ .collect(),
+ );
+ let batch = read_batch(&descriptors).await;
+ let payload = batch.column(1);
+ assert!(payload.is_null(1));
+ let (children, empty) = if map {
+ let map = payload.as_any().downcast_ref::<MapArray>().unwrap();
+ (map.value(4).column(1).clone(), map.value_length(2))
+ } else {
+ let list =
payload.as_any().downcast_ref::<ListArray>().unwrap();
+ (list.value(4), list.value_length(2))
+ };
+ assert_eq!(empty, 0);
+ let values = children
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap();
+ assert!(values.is_null(1));
+ assert_eq!(
+ BlobDescriptor::deserialize(values.value(0))
+ .unwrap()
+ .length(),
+ 3
+ );
+ assert_eq!(
+ BlobDescriptor::deserialize(values.value(2))
+ .unwrap()
+ .length(),
+ 0
+ );
+ }
+ }
+}
+
+async fn seed(table: &Table) {
+ seed_chunks(table, 5).await;
+}
+
+async fn seed_chunks(table: &Table, chunk_size: usize) {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ let batch = RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![0, 1, 2, 3, 4])) as ArrayRef,
+ ),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from(vec![
+ Some(b"old-0".as_slice()),
+ None,
+ Some(b"".as_slice()),
+ Some(b"old-3".as_slice()),
+ Some(b"old-4".as_slice()),
+ ])) as ArrayRef,
+ ),
+ (
+ "value",
+ Arc::new(Int32Array::from(vec![10, 11, 12, 13, 14])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ for offset in (0..batch.num_rows()).step_by(chunk_size) {
+ writer
+ .write_arrow_batch(&batch.slice(offset,
chunk_size.min(batch.num_rows() - offset)))
+ .await
+ .unwrap();
+ }
+ builder
+ .new_commit()
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+}
+
+#[tokio::test]
+async fn independently_rolled_blob_files_do_not_define_update_ranges() {
+ let table = table(&[
+ ("blob.target-file-size", "35 B"),
+ ("target-file-row-num", "100"),
+ ])
+ .await;
+ seed(&table).await;
+ let plan = table.new_read_builder().new_scan().plan().await.unwrap();
+ let normal_ranges = plan
+ .splits()
+ .iter()
+ .flat_map(|split| split.data_files())
+ .filter(|file| file.file_name.ends_with(".parquet"))
+ .map(|file| file.row_id_range().unwrap())
+ .collect::<Vec<_>>();
+ assert_eq!(normal_ranges, vec![(0, 4)]);
+ let blob_ranges = plan
+ .splits()
+ .iter()
+ .flat_map(|split| split.data_files())
+ .filter(|file| file.file_name.ends_with(".blob"))
+ .map(|file| file.row_id_range().unwrap())
+ .collect::<Vec<_>>();
+ assert!(
+ blob_ranges
+ .iter()
+ .any(|&(first, last)| first < 4 && last == 4),
+ "{blob_ranges:?}"
+ );
+ let builder = table.new_write_builder();
+ let mut update = builder.new_update().unwrap();
+ update
+ .with_update_type(vec!["payload".into(), "value".into()])
+ .unwrap();
+ let batch = RecordBatch::try_from_iter([
+ ("_ROW_ID", Arc::new(Int64Array::from(vec![4])) as ArrayRef),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from(vec![Some(b"new-4".as_slice())]))
as ArrayRef,
+ ),
+ ("value", Arc::new(Int32Array::from(vec![99])) as ArrayRef),
+ ])
+ .unwrap();
+ let messages = update
+ .update_by_arrow_with_row_id(vec![batch])
+ .await
+ .unwrap();
+ assert!(messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .all(|file| file.first_row_id == Some(0) && file.row_count == 5));
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(
+ rows(&table).await,
+ vec![
+ (0, Some(b"old-0".to_vec()), 10),
+ (1, None, 11),
+ (2, Some(vec![]), 12),
+ (3, Some(b"old-3".to_vec()), 13),
+ (4, Some(b"new-4".to_vec()), 99),
+ ]
+ );
+}
+
+async fn rows(table: &Table) -> Vec<(i32, Option<Vec<u8>>, i32)> {
+ let builder = table.new_read_builder();
+ 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 mut rows = Vec::new();
+ for batch in batches {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let blobs = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap();
+ let values = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ rows.extend(
+ ids.values()
+ .iter()
+ .zip(blobs.iter())
+ .zip(values.values())
+ .map(|((id, blob), value)| (*id, blob.map(Vec::from), *value)),
+ );
+ }
+ rows.sort_by_key(|row| row.0);
+ rows
+}
+
+#[tokio::test]
+async fn raw_blob_update_distinguishes_null_from_java_placeholder() {
+ let table = table(&[]).await;
+ seed(&table).await;
+ let builder = table.new_write_builder();
+ let mut update = builder.new_update().unwrap();
+ update.with_update_type(vec!["payload".into()]).unwrap();
+ let batch = RecordBatch::try_from_iter([
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from(vec![2, 4])) as ArrayRef,
+ ),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from(vec![
+ None,
+ Some(b"new-4".as_slice()),
+ ])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let messages = update
+ .update_by_arrow_with_row_id(vec![batch])
+ .await
+ .unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].check_from_snapshot, Some(1));
+ assert_eq!(messages[0].new_files.len(), 1);
+ let delta = &messages[0].new_files[0];
+ assert!(delta.file_name.ends_with(".blob"));
+ assert_eq!(delta.write_cols, Some(vec!["payload".into()]));
+ assert_eq!((delta.first_row_id, delta.row_count), (Some(0), 5));
+ assert_eq!(
+ (delta.min_sequence_number, delta.max_sequence_number),
+ (0, 0)
+ );
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(
+ rows(&table).await,
+ vec![
+ (0, Some(b"old-0".to_vec()), 10),
+ (1, None, 11),
+ (2, None, 12),
+ (3, Some(b"old-3".to_vec()), 13),
+ (4, Some(b"new-4".to_vec()), 14),
+ ]
+ );
+}
+
+async fn physical_files(table: &Table) -> HashSet<String> {
+ table
+ .file_io()
+ .list_status_recursive(table.location())
+ .await
+ .unwrap()
+ .into_iter()
+ .filter(|entry| entry.path.ends_with(".blob") ||
entry.path.ends_with(".parquet"))
+ .map(|entry| entry.path)
+ .collect()
+}
+
+#[tokio::test]
+async fn raw_blob_upsert_matches_keys_before_writing_payloads() {
+ let table = table(&[("blob.target-file-size", "20 B")]).await;
+ seed(&table).await;
+ let builder = table.new_write_builder();
+ let mut update = builder.new_update().unwrap();
+ update.with_update_type(vec!["payload".into()]).unwrap();
+ let source = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![1, 5, 5])) as ArrayRef),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from(vec![
+ Some(b"matched".as_slice()),
+ Some(b"shadowed"),
+ None,
+ ])) as ArrayRef,
+ ),
+ (
+ "value",
+ Arc::new(Int32Array::from(vec![111, 500, 501])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let messages = update
+ .upsert_by_arrow_with_key(vec![source], vec!["id".into()])
+ .await
+ .unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(
+ rows(&table).await,
+ vec![
+ (0, Some(b"old-0".to_vec()), 10),
+ (1, Some(b"matched".to_vec()), 11),
+ (2, Some(vec![]), 12),
+ (3, Some(b"old-3".to_vec()), 13),
+ (4, Some(b"old-4".to_vec()), 14),
+ (5, None, 501),
+ ]
+ );
+}
+
+#[tokio::test]
+async fn raw_blob_update_uses_baseline_metadata_without_opening_old_payloads()
{
+ let table = table(&[]).await;
+ seed(&table).await;
+ for path in physical_files(&table)
+ .await
+ .into_iter()
+ .filter(|path| path.ends_with(".blob"))
+ {
+ table.file_io().delete_file(&path).await.unwrap();
+ }
+ let messages = apply(&table, matched(vec![1], vec![Some(b"new")])).await;
+ assert_eq!(messages[0].new_files[0].row_count, 2);
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["id", "value"]).unwrap();
+ 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();
+ assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 5);
+}
+
+#[tokio::test]
+async fn abort_and_commit_conflicts_preserve_prepared_blob_files() {
+ for published in [false, true] {
+ let table = table(&[]).await;
+ seed(&table).await;
+ let builder = table.new_write_builder();
+ let mut updater = builder
+ .new_update()
+ .unwrap()
+ .new_update_by_row_id()
+ .await
+ .unwrap();
+ let messages = updater
+ .update_columns(
+ vec![matched(vec![1], vec![Some(b"new")])],
+ vec!["payload".into()],
+ )
+ .await
+ .unwrap();
+ let files = physical_files(&table).await;
+ if published {
+ builder.new_commit().commit(messages.clone()).await.unwrap();
+ }
+ updater.abort().await.unwrap();
+ updater.abort().await.unwrap();
+ assert_eq!(physical_files(&table).await, files);
+ if !published {
+ builder.new_commit().commit(messages).await.unwrap();
+ }
+ assert_eq!(rows(&table).await[1].1, Some(b"new".to_vec()));
+ }
+ let table = table(&[]).await;
+ seed(&table).await;
+ let stale = apply(&table, matched(vec![1], vec![Some(b"stale")])).await;
+ let latest = apply(&table, matched(vec![1], vec![Some(b"latest")])).await;
+ let files = physical_files(&table).await;
+ let builder = table.new_write_builder();
+ builder.new_commit().commit(latest).await.unwrap();
+ let error = builder.new_commit().commit(stale).await.unwrap_err();
+ assert!(error.to_string().contains("conflict"), "{error}");
+ assert_eq!(physical_files(&table).await, files);
+ assert_eq!(rows(&table).await[1].1, Some(b"latest".to_vec()));
+}
+
+#[tokio::test]
+async fn blob_update_keeps_deleted_rows_deleted_after_partial_rewrite() {
+ let table = table(&[]).await;
+ seed(&table).await;
+ let builder = table.new_write_builder();
+ let deletion = builder
+ .new_update()
+ .unwrap()
+ .delete_by_row_id(vec![0, 2])
+ .await
+ .unwrap();
+ builder.new_commit().commit(deletion).await.unwrap();
+ let messages = apply(&table, matched(vec![1], vec![Some(b"new")])).await;
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(
+ rows(&table).await,
+ vec![
+ (1, Some(b"new".to_vec()), 11),
+ (3, Some(b"old-3".to_vec()), 13),
+ (4, Some(b"old-4".to_vec()), 14),
+ ]
+ );
+}
+
+struct BlobSource {
+ opened: AtomicUsize,
+ closed: Arc<AtomicUsize>,
+}
+
+struct BlobSourceFactory(Arc<BlobSource>);
+
+impl paimon::io::UriReaderFactory for BlobSourceFactory {
+ fn create(&self, uri: &str) -> paimon::Result<Arc<dyn
paimon::io::UriReader>> {
+ assert_eq!(uri, "custom://update");
+ Ok(self.0.clone())
+ }
+}
+
+#[async_trait::async_trait]
+impl paimon::io::UriReader for BlobSource {
+ async fn new_input_stream(
+ &self,
+ _uri: &str,
+ ) -> paimon::Result<Box<dyn paimon::io::UriInputStream>> {
+ self.opened.fetch_add(1, Ordering::Relaxed);
+ Ok(Box::new(BlobSourceStream {
+ position: 0,
+ closed: self.closed.clone(),
+ }))
+ }
+}
+
+struct BlobSourceStream {
+ position: usize,
+ closed: Arc<AtomicUsize>,
+}
+
+#[async_trait::async_trait]
+impl paimon::io::UriInputStream for BlobSourceStream {
+ async fn read(&mut self, length: usize) -> paimon::Result<bytes::Bytes> {
+ assert!(length <= 4096);
+ let bytes = bytes::Bytes::from_static(b"streamed value");
+ let end = (self.position + length).min(bytes.len());
+ let result = bytes.slice(self.position..end);
+ self.position = end;
+ Ok(result)
+ }
+
+ async fn close(&mut self) -> paimon::Result<()> {
+ self.closed.fetch_add(1, Ordering::Relaxed);
+ Ok(())
+ }
+}
+
+#[tokio::test]
+async fn
row_id_blob_reader_factory_streams_unknown_lengths_and_closes_sources() {
+ let table = table(&[("blob.target-file-size", "20 B")]).await;
+ seed(&table).await;
+ let reader = Arc::new(BlobSource {
+ opened: AtomicUsize::new(0),
+ closed: Arc::new(AtomicUsize::new(0)),
+ });
+ let builder = table.new_write_builder();
+ let mut updater = builder
+ .new_update()
+ .unwrap()
+ .new_update_by_row_id()
+ .await
+ .unwrap();
+
updater.with_blob_uri_reader_factory(Some(Arc::new(BlobSourceFactory(reader.clone()))));
+ let descriptor = BlobDescriptor::new("custom://update".into(), 0,
-1).serialize();
+ let messages = updater
+ .update_columns(
+ vec![matched(
+ vec![1, 4],
+ vec![Some(&descriptor), Some(&descriptor)],
+ )],
+ vec!["payload".into()],
+ )
+ .await
+ .unwrap();
+ assert_eq!(reader.opened.load(Ordering::Relaxed), 2);
+ assert_eq!(reader.closed.load(Ordering::Relaxed), 2);
+ builder.new_commit().commit(messages).await.unwrap();
+ let rows = rows(&table).await;
+ assert_eq!(rows[1].1, Some(b"streamed value".to_vec()));
+ assert_eq!(rows[4].1, Some(b"streamed value".to_vec()));
+}
