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 f99d5260 feat(core): support scoped Blob sources in table writes and
updates (#1082)
f99d5260 is described below
commit f99d5260e3498c80a9f140c7341aaa32f7f03ce0
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 9 13:23:24 2026 +0800
feat(core): support scoped Blob sources in table writes and updates (#1082)
---
.../python/python/pypaimon_rust/datafusion.pyi | 2 +
bindings/python/src/blob_uri_reader.rs | 26 ++
bindings/python/src/write.rs | 52 ++-
bindings/python/tests/test_blob_uri_reader.py | 167 ++++++++
crates/paimon/src/arrow/format/blob/writer.rs | 6 +-
crates/paimon/src/io/uri_reader/custom.rs | 7 +
crates/paimon/src/table/data_evolution_blob.rs | 18 +-
crates/paimon/src/table/table_merge_into.rs | 3 +
crates/paimon/src/table/table_update.rs | 28 +-
crates/paimon/src/table/table_update_predicate.rs | 3 +
crates/paimon/src/table/table_upsert.rs | 35 +-
crates/paimon/src/table/table_write.rs | 32 +-
crates/paimon/src/table/write_batch_normalize.rs | 442 ++++++++++++++++++++-
crates/paimon/tests/blob_update_test.rs | 272 +++++++++++++
14 files changed, 1036 insertions(+), 57 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index af2b9f3b..0b701ded 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -287,6 +287,7 @@ class BatchTableWrite:
...
class BatchTableUpdate:
+ def _with_blob_uri_reader_factory(self, factory: Optional[Any]) -> None:
...
def merge_into(
self, source: Optional[Union[pyarrow.Table, Table]], *, on:
Sequence[Tuple[str, str]],
when_matched: Sequence[Dict[str, Any]], when_not_matched:
Sequence[Dict[str, Any]]
@@ -329,6 +330,7 @@ class TableUpdateByRowId:
def _abort(self) -> None: ...
class StreamTableUpdate:
+ def _with_blob_uri_reader_factory(self, factory: Optional[Any]) -> None:
...
def merge_into(
self, source: Optional[Union[pyarrow.Table, Table]], *, on:
Sequence[Tuple[str, str]],
when_matched: Sequence[Dict[str, Any]], when_not_matched:
Sequence[Dict[str, Any]],
diff --git a/bindings/python/src/blob_uri_reader.rs
b/bindings/python/src/blob_uri_reader.rs
index 46608317..39c21c16 100644
--- a/bindings/python/src/blob_uri_reader.rs
+++ b/bindings/python/src/blob_uri_reader.rs
@@ -80,6 +80,18 @@ pub(crate) struct PythonUriReaderFactory {
readers: Mutex<HashMap<usize, Weak<PythonUriReader>>>,
}
+pub(crate) fn optional_uri_reader_factory(
+ factory: Option<Py<PyAny>>,
+ py: Python<'_>,
+) -> PyResult<Option<Arc<dyn UriReaderFactory>>> {
+ factory
+ .map(|factory| {
+ PythonUriReaderFactory::new(factory, py)
+ .map(|factory| Arc::new(factory) as Arc<dyn UriReaderFactory>)
+ })
+ .transpose()
+}
+
impl PythonUriReaderFactory {
pub(crate) fn new(factory: Py<PyAny>, py: Python<'_>) -> PyResult<Self> {
if !factory.bind(py).getattr("create")?.is_callable() {
@@ -95,6 +107,20 @@ impl PythonUriReaderFactory {
}
impl UriReaderFactory for PythonUriReaderFactory {
+ fn supports_uri(&self, uri: &str) -> paimon::Result<bool> {
+ tokio::task::block_in_place(|| {
+ Python::attach(|py| {
+ let factory = self.factory.bind(py);
+ if factory.hasattr("_supports_uri")? {
+ factory.call_method1("_supports_uri", (uri,))?.extract()
+ } else {
+ Ok(true)
+ }
+ })
+ })
+ .map_err(callback_error)
+ }
+
fn create(&self, uri: &str) -> paimon::Result<Arc<dyn UriReader>> {
tokio::task::block_in_place(|| {
Python::attach(|py| {
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index eec73110..da7d6b9b 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -31,7 +31,7 @@ use pyo3::exceptions::{PyRuntimeError, PyTypeError,
PyValueError};
use pyo3::prelude::*;
use pyo3::types::{PyBytes, PyDict, PyString};
-use crate::blob_uri_reader::{callback_error, to_callback_error,
PythonUriReaderFactory};
+use crate::blob_uri_reader::{callback_error, optional_uri_reader_factory,
to_callback_error};
use crate::error::to_py_err;
use crate::predicate::{dict_to_table_predicate, py_to_datum};
@@ -568,6 +568,16 @@ struct UpdateContext {
}
impl UpdateContext {
+ fn with_blob_uri_reader_factory(
+ &mut self,
+ factory: Option<Py<PyAny>>,
+ py: Python<'_>,
+ ) -> PyResult<()> {
+ self.inner
+ .with_blob_uri_reader_factory(optional_uri_reader_factory(factory,
py)?);
+ Ok(())
+ }
+
fn new(context: &WriteContext) -> PyResult<Self> {
Ok(Self {
inner: context
@@ -631,7 +641,7 @@ impl UpdateContext {
return Err(error);
}
Ok(wrap_messages(
- result.map_err(to_py_err)?,
+ result.map_err(to_callback_error)?,
&self.table_location,
&self.commit_user,
))
@@ -665,7 +675,7 @@ impl UpdateContext {
return Err(error);
}
Ok(wrap_messages(
- result.map_err(to_py_err)?,
+ result.map_err(to_callback_error)?,
&self.table_location,
&self.commit_user,
))
@@ -679,7 +689,7 @@ impl UpdateContext {
let batches = arrow_table_batches(table)?;
let messages = py
.detach(||
runtime().block_on(self.inner.update_by_arrow_with_row_id(batches)))
- .map_err(to_py_err)?;
+ .map_err(to_callback_error)?;
Ok(wrap_messages(
messages,
&self.table_location,
@@ -720,7 +730,7 @@ impl UpdateContext {
return Err(error);
}
Ok(wrap_messages(
- result.map_err(to_py_err)?,
+ result.map_err(to_callback_error)?,
&self.table_location,
&self.commit_user,
))
@@ -746,7 +756,7 @@ impl UpdateContext {
let batches = arrow_upsert_batches(input)?;
let messages = py
.detach(||
runtime().block_on(self.inner.upsert_by_arrow_with_key(batches, keys)))
- .map_err(to_py_err)?;
+ .map_err(to_callback_error)?;
Ok(wrap_messages(
messages,
&self.table_location,
@@ -776,12 +786,7 @@ impl WriteState {
factory: Option<Py<PyAny>>,
py: Python<'_>,
) -> PyResult<()> {
- let factory = factory
- .map(|factory| {
- PythonUriReaderFactory::new(factory, py)
- .map(|factory| Arc::new(factory) as Arc<dyn
paimon::io::UriReaderFactory>)
- })
- .transpose()?;
+ let factory = optional_uri_reader_factory(factory, py)?;
self.inner
.as_mut()
.ok_or_else(|| PyRuntimeError::new_err("TableWrite is closed"))?
@@ -882,12 +887,7 @@ impl PyTableUpdateByRowId {
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()?;
+ let factory = optional_uri_reader_factory(factory, py)?;
self.inner.with_blob_uri_reader_factory(factory);
Ok(())
}
@@ -931,6 +931,14 @@ pub struct PyStreamTableUpdate {
#[pymethods]
impl PyStreamTableUpdate {
+ fn _with_blob_uri_reader_factory(
+ &mut self,
+ py: Python<'_>,
+ factory: Option<Py<PyAny>>,
+ ) -> PyResult<()> {
+ self.context.with_blob_uri_reader_factory(factory, py)
+ }
+
#[pyo3(signature = (source, *, on, when_matched, when_not_matched,
commit_identifier))]
fn merge_into(
&self,
@@ -1018,6 +1026,14 @@ impl PyStreamTableUpdate {
#[pymethods]
impl PyBatchTableUpdate {
+ fn _with_blob_uri_reader_factory(
+ &mut self,
+ py: Python<'_>,
+ factory: Option<Py<PyAny>>,
+ ) -> PyResult<()> {
+ self.context.with_blob_uri_reader_factory(factory, py)
+ }
+
#[pyo3(signature = (source, *, on, when_matched, when_not_matched))]
fn merge_into(
&self,
diff --git a/bindings/python/tests/test_blob_uri_reader.py
b/bindings/python/tests/test_blob_uri_reader.py
index 0e20ed09..6c0fbaac 100644
--- a/bindings/python/tests/test_blob_uri_reader.py
+++ b/bindings/python/tests/test_blob_uri_reader.py
@@ -22,6 +22,7 @@ import pyarrow as pa
import pytest
from test_blob_consumer import _table
+from pypaimon_rust.datafusion import PaimonCatalog, SQLContext
def descriptor(offset, length):
@@ -105,3 +106,169 @@ def
test_custom_uri_read_error_preserves_original_exception_and_closes_once(tmp_
finally:
writer.close()
assert not list(tmp_path.rglob('*.blob'))
+
+
+def _update_table(tmp_path):
+ context = SQLContext()
+ context.register_catalog('paimon', {'warehouse': str(tmp_path)})
+ context.sql('CREATE SCHEMA paimon.blobs')
+ context.sql("""CREATE TABLE paimon.blobs.t (id INT, payload BLOB) WITH (
+ 'row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true',
+ 'blob.copy-buffer-size' = '2 B')""")
+ return context, PaimonCatalog({'warehouse':
str(tmp_path)}).get_table('blobs.t')
+
+
+def _reference(uri, offset=0, length=-1):
+ encoded = uri.encode('utf-8')
+ return (struct.pack('<BQi', 2, 0x424C4F4244455343, len(encoded)) + encoded
+ + struct.pack('<qq', offset, length))
+
+
+def _batch(ids, values):
+ return pa.record_batch([pa.array(ids, type=pa.int32()), pa.array(values,
type=pa.large_binary())],
+ names=['id', 'payload'])
+
+
+def _read(context):
+ return pa.Table.from_batches(context.sql('SELECT id, payload FROM
paimon.blobs.t ORDER BY id')).to_pydict()
+
+
[email protected]('stream', [False, True])
+def test_scoped_blob_factory_keeps_native_file_references(tmp_path, stream):
+ context, table = _update_table(tmp_path)
+ source = tmp_path / 'source'
+ source.write_bytes(b'prefixREFERENCEDsuffix')
+ streams = []
+ selected = []
+
+ class Reader:
+ def new_input_stream(self, uri):
+ assert uri == 'custom://object'
+ value = io.BytesIO(b'OBJECT')
+ streams.append(value)
+ return value
+
+ class Factory:
+ def _supports_uri(self, uri):
+ selected.append(uri)
+ return uri.startswith('custom://')
+
+ def create(self, uri):
+ assert uri == 'custom://object', 'Native references must not be
delegated to Python'
+ return Reader()
+
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.with_blob_uri_reader_factory(Factory())
+ writer.write_arrow(_batch([0, 1], [_reference('custom://object'),
_reference(source.as_uri(), 6, 10)]))
+ assert len(streams) == 1 and streams[0].closed
+ messages = writer.prepare_commit(True, 7) if stream else
writer.prepare_commit()
+ commit = builder.new_commit()
+ commit.commit(7, messages) if stream else commit.commit(messages)
+ assert selected == ['custom://object', source.as_uri()]
+ assert _read(context) == {'id': [0, 1], 'payload': [b'OBJECT',
b'REFERENCED']}
+ finally:
+ writer.close()
+
+
[email protected]('stream, operation', [
+ (False, 'merge'), (True, 'merge'), (False, 'row_id'), (True, 'row_id'),
+ (False, 'grouped_row_id'), (False, 'incremental'), (True, 'incremental'),
+ (False, 'upsert'), (True, 'upsert'),
+ (False, 'predicate_scalar'), (True, 'predicate_scalar'),
+ (False, 'predicate_array'), (True, 'predicate_array'),
+])
[email protected]('failure_stage', ['select', 'create'])
+def test_update_factory_errors_keep_identity_and_existing_data(tmp_path,
stream, operation, failure_stage):
+ context, table = _update_table(tmp_path)
+ seed_builder = table.new_batch_write_builder()
+ seed = seed_builder.new_write()
+ try:
+ seed.write_arrow(_batch([0], [b'old']))
+ seed_builder.new_commit().commit(seed.prepare_commit())
+ finally:
+ seed.close()
+ row_id = context.sql('SELECT "_ROW_ID" FROM
paimon.blobs.t')[0].column(0)[0].as_py()
+ error = OSError('original factory failure')
+ calls = []
+ source_file = tmp_path / 'source'
+ source_file.write_bytes(b'native-readable')
+
+ class Factory:
+ def _supports_uri(self, uri):
+ calls.append('select')
+ if failure_stage == 'select':
+ raise error
+ return True
+
+ def create(self, uri):
+ calls.append('create')
+ if failure_stage == 'select':
+ pytest.fail('Selection errors must not fall back to any
reader')
+ raise error
+
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update()
+ update._with_blob_uri_reader_factory(Factory())
+ reference = _reference(source_file.as_uri())
+ source = pa.Table.from_batches([_batch([0], [reference])])
+ data = pa.table({'_ROW_ID': [row_id], 'payload': pa.array([reference],
type=pa.large_binary())})
+ extra = {'commit_identifier': 7} if stream else {}
+ with pytest.raises(OSError) as caught:
+ if operation == 'merge':
+ update.merge_into(source, on=[('id', 'id')], when_matched=[{
+ 'delete': False, 'assignments': [('payload', 'source',
'payload')],
+ }], when_not_matched=[], **extra)
+ elif operation == 'row_id':
+ update.update_by_arrow_with_row_id(data, **extra)
+ elif operation == 'grouped_row_id':
+ update.update_by_arrow_batches_with_row_id([data], **extra)
+ elif operation == 'incremental':
+ update.new_update_by_row_id(**extra).update_columns(data,
['payload'])
+ elif operation.startswith('predicate_'):
+ value = (reference if operation == 'predicate_scalar'
+ else pa.array([reference], type=pa.large_binary()))
+ update.update_by_predicate(None, {'payload': value}, **extra)
+ else:
+
update.with_update_type(['payload']).upsert_by_arrow_with_key(source, ['id'],
**extra)
+ assert caught.value is error
+ assert calls == (['select'] if failure_stage == 'select' else ['select',
'create'])
+ assert _read(context) == {'id': [0], 'payload': [b'old']}
+
+
[email protected]('stream', [False, True])
[email protected]('assignment', ['scalar', 'array', 'callable'])
+def
test_predicate_updates_share_blob_factory_across_matched_file_groups(tmp_path,
stream, assignment):
+ context, table = _update_table(tmp_path)
+ for ids, values in [([0, 1], [b'old-0', b'old-1']), ([2], [b'old-2'])]:
+ builder = table.new_batch_write_builder()
+ seed = builder.new_write()
+ try:
+ seed.write_arrow(_batch(ids, values))
+ builder.new_commit().commit(seed.prepare_commit())
+ finally:
+ seed.close()
+ factory = ReaderFactory()
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update()
+ update._with_blob_uri_reader_factory(factory)
+ reference = descriptor(2, -1)
+ read_columns = None
+ if assignment == 'scalar':
+ value = reference
+ elif assignment == 'array':
+ value = pa.array([reference, reference], type=pa.large_binary())
+ else:
+ def value(rows):
+ assert rows.column_names == ['id', '_ROW_ID']
+ return pa.array([reference] * rows.num_rows,
type=pa.large_binary())
+ read_columns = ['id']
+ extra = {'commit_identifier': 7} if stream else {}
+ messages = update.update_by_predicate(
+ {'method': 'greaterOrEqual', 'field': 'id', 'literals': [1]},
+ {'payload': value}, read_columns=read_columns, **extra)
+ assert factory.opens == factory.closes == 2
+ commit = builder.new_commit()
+ commit.commit(7, messages) if stream else commit.commit(messages)
+ assert _read(context) == {'id': [0, 1, 2], 'payload': [b'old-0',
b'23456789', b'23456789']}
diff --git a/crates/paimon/src/arrow/format/blob/writer.rs
b/crates/paimon/src/arrow/format/blob/writer.rs
index 197550f1..c51e33e4 100644
--- a/crates/paimon/src/arrow/format/blob/writer.rs
+++ b/crates/paimon/src/arrow/format/blob/writer.rs
@@ -303,7 +303,11 @@ impl BlobFormatWriter {
}
let descriptor = BlobDescriptor::deserialize(value)?;
let range = descriptor.range_spec()?;
- if let Some(factory) = self.uri_reader_factory.clone() {
+ let factory = match &self.uri_reader_factory {
+ Some(factory) if factory.supports_uri(descriptor.uri())? =>
Some(factory.clone()),
+ _ => None,
+ };
+ if let Some(factory) = factory {
return self
.copy_custom_reference(factory.as_ref(), &descriptor, hasher)
.await;
diff --git a/crates/paimon/src/io/uri_reader/custom.rs
b/crates/paimon/src/io/uri_reader/custom.rs
index 264b25fc..eee872de 100644
--- a/crates/paimon/src/io/uri_reader/custom.rs
+++ b/crates/paimon/src/io/uri_reader/custom.rs
@@ -24,6 +24,13 @@ use std::sync::Arc;
/// Select a URI reader, corresponding to Java's UriReaderFactory.create.
/// Return the same Arc when references share the same reader identity.
pub trait UriReaderFactory: Send + Sync {
+ /// Limit an application reader to the references it owns. Other URIs use
+ /// the table's native FileIO/HTTP reader. Existing factories own all URIs.
+ /// Errors are propagated; they never select a different reader.
+ fn supports_uri(&self, _uri: &str) -> Result<bool> {
+ Ok(true)
+ }
+
fn create(&self, uri: &str) -> Result<Arc<dyn UriReader>>;
}
diff --git a/crates/paimon/src/table/data_evolution_blob.rs
b/crates/paimon/src/table/data_evolution_blob.rs
index ea347aef..0ddacd59 100644
--- a/crates/paimon/src/table/data_evolution_blob.rs
+++ b/crates/paimon/src/table/data_evolution_blob.rs
@@ -223,7 +223,7 @@ pub(super) fn delta_batch(
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 {
+ let DataType::Map(_, _) = &target else {
return Ok(target);
};
let list_input = batches
@@ -238,16 +238,8 @@ fn update_arrow_type(field: &DataField, batches:
&[RecordBatch]) -> Result<arrow
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,
- ))))
+ Ok(
+ super::super::write_batch_normalize::blob_map_row_type(&target)
+ .expect("logical Blob maps have key/value entries"),
+ )
}
diff --git a/crates/paimon/src/table/table_merge_into.rs
b/crates/paimon/src/table/table_merge_into.rs
index 47ae2147..a37ed906 100644
--- a/crates/paimon/src/table/table_merge_into.rs
+++ b/crates/paimon/src/table/table_merge_into.rs
@@ -210,6 +210,7 @@ pub(super) async fn merge_into(
on: Vec<(String, String)>,
matched: Vec<WhenMatched>,
not_matched: Vec<WhenNotMatched>,
+ blob_uri_reader_factory: Option<Arc<dyn crate::io::UriReaderFactory>>,
) -> crate::Result<Vec<CommitMessage>> {
if on.is_empty() || matched.is_empty() && not_matched.is_empty() {
return Err(invalid("MERGE requires ON keys and at least one action"));
@@ -597,6 +598,7 @@ pub(super) async fn merge_into(
let mut messages = Vec::new();
if !updates.is_empty() {
let mut updater = TableUpdateByRowId::with_index(table, index)?;
+ updater.with_blob_uri_reader_factory(blob_uri_reader_factory.clone());
messages.extend(updater.update_columns(updates,
update_columns).await?);
}
if !deletes.is_empty() {
@@ -609,6 +611,7 @@ pub(super) async fn merge_into(
.new_write_builder()
.with_commit_user(commit_user.to_string())?
.new_write()?;
+ writer.with_blob_uri_reader_factory(blob_uri_reader_factory.clone())?;
let result = async {
for batch in inserts {
writer.write_arrow_batch(&batch).await?;
diff --git a/crates/paimon/src/table/table_update.rs
b/crates/paimon/src/table/table_update.rs
index 012fa056..a516fb70 100644
--- a/crates/paimon/src/table/table_update.rs
+++ b/crates/paimon/src/table/table_update.rs
@@ -50,6 +50,7 @@ pub struct TableUpdate {
table: Table,
commit_user: String,
update_cols: Option<Vec<String>>,
+ blob_uri_reader_factory: Option<std::sync::Arc<dyn
crate::io::UriReaderFactory>>,
}
impl TableUpdate {
@@ -58,9 +59,20 @@ impl TableUpdate {
table: table.clone(),
commit_user,
update_cols: None,
+ blob_uri_reader_factory: None,
}
}
+ /// Configure readers for Blob input references. The same factory supplies
+ /// matched updates and appended rows in upsert and MERGE operations.
+ 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
+ }
+
/// Select columns to update when a key or row ID matches. Duplicate names
/// are removed in first-occurrence order, as in PyPaimon.
pub fn with_update_type(&mut self, update_cols: Vec<String>) ->
crate::Result<&mut Self> {
@@ -125,7 +137,9 @@ impl TableUpdate {
/// Create a row-ID updater sharing one snapshot across per-call columns.
pub async fn new_update_by_row_id(&self) ->
crate::Result<TableUpdateByRowId> {
- TableUpdateByRowId::new(&self.table).await
+ let mut writer = TableUpdateByRowId::new(&self.table).await?;
+
writer.with_blob_uri_reader_factory(self.blob_uri_reader_factory.clone());
+ Ok(writer)
}
/// Update existing rows from chunks of one Arrow table containing
`_ROW_ID`.
@@ -186,8 +200,14 @@ impl TableUpdate {
assignments: Vec<(String, super::UpdateAssignment)>,
read_columns: Vec<String>,
) -> crate::Result<Vec<CommitMessage>> {
- super::table_update_predicate::update(&self.table, predicate,
assignments, read_columns)
- .await
+ super::table_update_predicate::update(
+ &self.table,
+ predicate,
+ assignments,
+ read_columns,
+ self.blob_uri_reader_factory.clone(),
+ )
+ .await
}
/// Upsert Arrow rows by composite key through the core upsert
@@ -212,6 +232,7 @@ impl TableUpdate {
.filter(|columns| !columns.is_empty())
.unwrap_or_else(|| self.all_fields()),
)?;
+
writer.with_blob_uri_reader_factory(self.blob_uri_reader_factory.clone());
for batch in batches {
writer.add_batch(batch)?;
}
@@ -234,6 +255,7 @@ impl TableUpdate {
on,
when_matched,
when_not_matched,
+ self.blob_uri_reader_factory.clone(),
)
.await
}
diff --git a/crates/paimon/src/table/table_update_predicate.rs
b/crates/paimon/src/table/table_update_predicate.rs
index 8bf5ef8b..675db998 100644
--- a/crates/paimon/src/table/table_update_predicate.rs
+++ b/crates/paimon/src/table/table_update_predicate.rs
@@ -18,6 +18,7 @@
//! Predicate matching and assignment orchestration for TableUpdate.
use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
use arrow_array::RecordBatch;
use futures::TryStreamExt;
@@ -104,6 +105,7 @@ pub(super) async fn update(
predicate: Option<Predicate>,
assignments: Vec<(String, UpdateAssignment)>,
read_columns: Vec<String>,
+ blob_uri_reader_factory: Option<Arc<dyn crate::io::UriReaderFactory>>,
) -> crate::Result<Vec<CommitMessage>> {
let combine_all = validate(table, &assignments, &read_columns)?;
let Some(snapshot) = super::time_travel::resolve_snapshot(table).await?
else {
@@ -135,6 +137,7 @@ pub(super) async fn update(
let index = RowIdFileIndex::from_splits(scan_table.clone(),
plan.splits())?;
let groups = ordered_file_groups(plan.splits())?;
let mut updater = TableUpdateByRowId::with_index(table, index)?;
+ updater.with_blob_uri_reader_factory(blob_uri_reader_factory);
let reader = read_builder.new_read()?;
let schema =
crate::arrow::build_target_arrow_schema(table.schema().fields())?;
let columns: Vec<_> = assignments.iter().map(|(name, _)|
name.clone()).collect();
diff --git a/crates/paimon/src/table/table_upsert.rs
b/crates/paimon/src/table/table_upsert.rs
index 0ce5aa75..769a9734 100644
--- a/crates/paimon/src/table/table_upsert.rs
+++ b/crates/paimon/src/table/table_upsert.rs
@@ -30,7 +30,7 @@ use indexmap::IndexMap;
use super::update_input::unique_column_index;
use super::upsert_key_matcher::UpsertKeyMatcher;
-use super::write_batch_normalize::normalize_write_array;
+use super::write_batch_normalize::{normalize_blob_write_field,
normalize_write_array};
use crate::spec::{
batch_to_serialized_bytes, extract_datum, BinaryRow, BinaryRowBuilder,
CoreOptions, DataField,
};
@@ -87,6 +87,7 @@ pub(super) struct TableUpsert {
keys: Vec<String>,
update_columns: Vec<String>,
source: Vec<RecordBatch>,
+ blob_uri_reader_factory: Option<Arc<dyn crate::io::UriReaderFactory>>,
}
impl TableUpsert {
@@ -137,9 +138,17 @@ impl TableUpsert {
keys,
update_columns,
source: Vec::new(),
+ 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;
+ }
+
/// Add rows containing the keys and columns to write. Column order may
/// differ from the table schema; names and Arrow layouts follow the same
/// normalization as ordinary writes. Batches form one logical upsert
input,
@@ -161,16 +170,24 @@ impl TableUpsert {
}
let mut columns = Vec::with_capacity(target.fields().len());
let mut fields = Vec::new();
- for field in target.fields() {
+ let inline =
CoreOptions::new(self.table.schema().options()).blob_inline_fields();
+ for (index, field) in target.fields().iter().enumerate() {
let Some(column) = batch.column_by_name(field.name()) else {
continue;
};
- columns.push(
- normalize_write_array(column,
field.data_type()).map_err(|error| {
- invalid(format!("Invalid upsert column '{}': {error}",
field.name()))
- })?,
- );
- fields.push(field.clone());
+ let logical = &self.table.schema().fields()[index];
+ let normalized =
+ if logical.data_type().is_blob_file_field() &&
!inline.contains(logical.name()) {
+ normalize_blob_write_field(column, field)
+ } else {
+ normalize_write_array(column, field.data_type())
+ .map(|column| (field.as_ref().clone(), column))
+ };
+ let (field, column) = normalized.map_err(|error| {
+ invalid(format!("Invalid upsert column '{}': {error}",
field.name()))
+ })?;
+ columns.push(column);
+ fields.push(Arc::new(field));
}
let ordered = RecordBatch::try_new(Arc::new(Schema::new(fields)),
columns)
.map_err(|error| invalid(format!("cannot order upsert columns:
{error}")))?;
@@ -335,6 +352,7 @@ impl TableUpsert {
&self.table,
self.update_columns.clone(),
)?;
+
update.with_blob_uri_reader_factory(self.blob_uri_reader_factory.clone());
if let Some(snapshot_id) = plan.snapshot_id() {
update.pin_read_snapshot(snapshot_id);
}
@@ -357,6 +375,7 @@ impl TableUpsert {
.new_write_builder()
.with_commit_user(self.commit_user.clone())?
.new_write()?;
+
append.with_blob_uri_reader_factory(self.blob_uri_reader_factory.clone())?;
append.with_write_type(
group[0]
.schema()
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index e2a8e203..f4f055a8 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -45,7 +45,7 @@ use crate::table::partition_filter::PartitionFilter;
use crate::table::postpone_file_writer::{PostponeFileWriter,
PostponeWriteConfig};
use crate::table::prepared_files::PreparedFiles;
use crate::table::row_kind_generator::RowKindGenerator;
-use crate::table::write_batch_normalize::normalize_write_array;
+use crate::table::write_batch_normalize::{normalize_blob_write_field,
normalize_write_array};
use crate::table::{Snapshot, Table, TableScan};
use crate::Result;
use arrow_array::{Array, ArrayRef, Int8Array, RecordBatch};
@@ -938,10 +938,22 @@ impl TableWrite {
}
}
+ let blob_maps: HashSet<_> = self.table.schema().fields().iter()
+ .filter(|field| self.primary_key_indices.is_empty()
+ && matches!(field.data_type(), DataType::Map(map) if
map.value_type().is_blob_type())
+ && !self.blob_inline_fields.contains(field.name()))
+ .map(|field| field.name()).collect();
let mut normalized_columns: Vec<ArrayRef> =
Vec::with_capacity(actual_field_count);
+ let mut normalized_fields = Vec::with_capacity(actual_field_count);
for (index, expected_field) in
expected_schema.fields().iter().enumerate() {
let actual_field = actual_schema.field(index);
- let column = normalize_write_array(batch.column(index),
expected_field.data_type())
+ let normalized = if
blob_maps.contains(expected_field.name().as_str()) {
+ normalize_blob_write_field(batch.column(index), expected_field)
+ } else {
+ normalize_write_array(batch.column(index),
expected_field.data_type())
+ .map(|column| (expected_field.as_ref().clone(), column))
+ };
+ let (field, column) = normalized
.map_err(|error| crate::Error::DataInvalid {
message: format!(
"write batch schema data type mismatch for field '{}'
at index {index}: expected {:?}, actual {:?}: {error}",
@@ -951,6 +963,7 @@ impl TableWrite {
),
source: Some(Box::new(error)),
})?;
+ normalized_fields.push(Arc::new(field));
normalized_columns.push(column);
}
if includes_value_kind {
@@ -966,18 +979,13 @@ impl TableWrite {
});
}
normalized_columns.push(batch.column(table_field_count).clone());
+
normalized_fields.push(actual_schema.fields()[table_field_count].clone());
}
- let schema = if includes_value_kind {
- let mut fields =
expected_schema.fields().iter().cloned().collect::<Vec<_>>();
- fields.push(actual_schema.fields()[table_field_count].clone());
- Arc::new(arrow_schema::Schema::new_with_metadata(
- fields,
- expected_schema.metadata().clone(),
- ))
- } else {
- expected_schema.clone()
- };
+ let schema = Arc::new(arrow_schema::Schema::new_with_metadata(
+ normalized_fields,
+ expected_schema.metadata().clone(),
+ ));
RecordBatch::try_new(schema, normalized_columns).map_err(|error| {
crate::Error::DataInvalid {
message: format!("Failed to normalize write batch schema:
{error}"),
diff --git a/crates/paimon/src/table/write_batch_normalize.rs
b/crates/paimon/src/table/write_batch_normalize.rs
index 38d44086..0156133c 100644
--- a/crates/paimon/src/table/write_batch_normalize.rs
+++ b/crates/paimon/src/table/write_batch_normalize.rs
@@ -25,16 +25,188 @@
use crate::{Error, Result};
use arrow_array::{
- Array, ArrayRef, BinaryArray, FixedSizeBinaryArray, ListArray, MapArray,
StructArray,
+ Array, ArrayRef, BinaryArray, Decimal128Array, FixedSizeBinaryArray,
ListArray, MapArray,
+ StringArray, StructArray,
};
use arrow_buffer::{NullBuffer, OffsetBuffer, ScalarBuffer};
-use arrow_schema::DataType;
+use arrow_schema::{DataType, Field};
use std::sync::Arc;
pub(super) fn normalize_write_array(array: &ArrayRef, expected: &DataType) ->
Result<ArrayRef> {
normalize_visible_array(array, expected, None)
}
+/// Java Blob maps allow a NULL key. Arrow Map does not, so dedicated Blob
+/// writers use a list of key/value structs as their physical input. This
+/// representation must never be passed to an ordinary Parquet MAP writer.
+pub(super) fn blob_map_row_type(expected: &DataType) -> Option<DataType> {
+ let DataType::Map(entries, _) = expected else {
+ return None;
+ };
+ let DataType::Struct(fields) = entries.data_type() else {
+ return None;
+ };
+ if fields.len() != 2 {
+ return None;
+ }
+ let fields = vec![
+ Arc::new(fields[0].as_ref().clone().with_nullable(true)),
+ fields[1].clone(),
+ ];
+ Some(DataType::List(Arc::new(Field::new(
+ "item",
+ DataType::Struct(fields.into()),
+ false,
+ ))))
+}
+
+pub(super) fn normalize_blob_write_field(
+ array: &ArrayRef,
+ expected: &Field,
+) -> Result<(Field, ArrayRef)> {
+ let Some(target) = blob_map_row_type(expected.data_type()) else {
+ return Ok((
+ expected.clone(),
+ normalize_write_array(array, expected.data_type())?,
+ ));
+ };
+ let (entries, offsets, nulls) = match array.data_type() {
+ DataType::Map(_, _) => {
+ let map = array.as_any().downcast_ref::<MapArray>().unwrap();
+ (
+ Arc::new(map.entries().clone()) as ArrayRef,
+ map.offsets(),
+ map.nulls(),
+ )
+ }
+ DataType::List(_) => {
+ let list = array.as_any().downcast_ref::<ListArray>().unwrap();
+ (list.values().clone(), list.offsets(), list.nulls())
+ }
+ _ => {
+ return Err(Error::DataInvalid {
+ message: "MAP<X, BLOB> input must be a map or a list of
key/value structs".into(),
+ source: None,
+ })
+ }
+ };
+ let (entries, offsets) = visible_children(&entries, offsets, nulls)?;
+ let entries = entries
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .filter(|entries| entries.num_columns() == 2)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "MAP<X, BLOB> entries must contain exactly a key and a
value".into(),
+ source: None,
+ })?;
+ if entries.null_count() != 0 {
+ return Err(Error::DataInvalid {
+ message: "MAP<X, BLOB> entries cannot be NULL".into(),
+ source: None,
+ });
+ }
+ let DataType::List(child) = &target else {
+ unreachable!()
+ };
+ let DataType::Struct(fields) = child.data_type() else {
+ unreachable!()
+ };
+ // Key/value roles are positional, like ordinary Arrow MAP aliases.
+ let columns = entries
+ .columns()
+ .iter()
+ .zip(fields)
+ .enumerate()
+ .map(|(index, (column, field))| {
+ if index == 0 {
+ normalize_blob_map_key(column, field.data_type())
+ } else {
+ normalize_write_array(column, field.data_type())
+ }
+ })
+ .collect::<Result<Vec<_>>>()?;
+ let entries =
+ StructArray::try_new(fields.clone(), columns,
None).map_err(normalization_error)?;
+ let list = ListArray::try_new(child.clone(), offsets, Arc::new(entries),
nulls.cloned())
+ .map_err(normalization_error)?;
+ Ok((expected.clone().with_data_type(target), Arc::new(list)))
+}
+
+fn normalize_blob_map_key(array: &ArrayRef, expected: &DataType) ->
Result<ArrayRef> {
+ // Row bridges transport arbitrary-scale decimals as plain text. Arrow's
+ // declared-scale builder rejects these before Java HALF_UP can be applied.
+ if let (DataType::Utf8, DataType::Decimal128(precision, scale)) =
(array.data_type(), expected)
+ {
+ let strings = array.as_any().downcast_ref::<StringArray>().unwrap();
+ let values = strings
+ .iter()
+ .map(|value| {
+ value
+ .map(|text| {
+ decimal_blob_key(text, *precision,
*scale).ok_or_else(|| {
+ Error::DataInvalid {
+ message: "MAP DECIMAL key is invalid or
exceeds declared precision"
+ .into(),
+ source: None,
+ }
+ })
+ })
+ .transpose()
+ })
+ .collect::<Result<Vec<_>>>()?;
+ return Ok(Arc::new(
+ Decimal128Array::from(values)
+ .with_precision_and_scale(*precision, *scale)
+ .map_err(normalization_error)?,
+ ));
+ }
+ normalize_write_array(array, expected)
+}
+
+/// Decimal.fromBigDecimal: round magnitude HALF_UP, then check precision.
+/// Discarded fractional digits are never accumulated into i128, so source
+/// decimals can have more digits than the target's maximum precision of 38.
+fn decimal_blob_key(text: &str, precision: u8, scale: i8) -> Option<i128> {
+ let scale = usize::try_from(scale).ok()?;
+ let (negative, text) = match text.strip_prefix('-') {
+ Some(text) => (true, text),
+ None => (false, text.strip_prefix('+').unwrap_or(text)),
+ };
+ let (whole, fraction) = text.split_once('.').unwrap_or((text, ""));
+ if whole.is_empty() && fraction.is_empty()
+ || !whole
+ .bytes()
+ .chain(fraction.bytes())
+ .all(|byte| byte.is_ascii_digit())
+ {
+ return None;
+ }
+ let whole = whole.trim_start_matches('0');
+ if whole.len() > usize::from(precision) {
+ return None;
+ }
+ let mut value = 0i128;
+ for digit in whole.bytes().chain(fraction.bytes().take(scale)) {
+ value = value
+ .checked_mul(10)?
+ .checked_add(i128::from(digit - b'0'))?;
+ }
+ for _ in fraction.len()..scale {
+ value = value.checked_mul(10)?;
+ }
+ if fraction
+ .as_bytes()
+ .get(scale)
+ .is_some_and(|digit| *digit >= b'5')
+ {
+ value = value.checked_add(1)?;
+ }
+ if value.checked_ilog10().map_or(1, |log| log + 1) > u32::from(precision) {
+ return None;
+ }
+ Some(if negative { -value } else { value })
+}
+
fn normalize_visible_array(
array: &ArrayRef,
expected: &DataType,
@@ -224,6 +396,272 @@ mod tests {
use arrow_buffer::{OffsetBuffer, ScalarBuffer};
use arrow_schema::Field;
+ fn blob_map_field(nullable_value: bool) -> Field {
+ Field::new(
+ "payload",
+ DataType::Map(
+ Arc::new(Field::new(
+ "entries",
+ DataType::Struct(
+ vec![
+ Field::new("key", DataType::Utf8, false),
+ Field::new("value", DataType::LargeBinary,
nullable_value),
+ ]
+ .into(),
+ ),
+ false,
+ )),
+ false,
+ ),
+ true,
+ )
+ }
+
+ fn blob_map_rows(
+ keys: Vec<Option<&str>>,
+ values: Vec<Option<&[u8]>>,
+ offsets: Vec<i32>,
+ nulls: Option<NullBuffer>,
+ entry_nulls: Option<NullBuffer>,
+ ) -> ArrayRef {
+ let entries = StructArray::new(
+ vec![
+ Field::new("source_key", DataType::Utf8, true),
+ Field::new("source_value", DataType::LargeBinary, true),
+ ]
+ .into(),
+ vec![
+ Arc::new(StringArray::from(keys)),
+ Arc::new(LargeBinaryArray::from(values)),
+ ],
+ entry_nulls,
+ );
+ Arc::new(ListArray::new(
+ Arc::new(Field::new("input", entries.data_type().clone(), true)),
+ OffsetBuffer::new(ScalarBuffer::from(offsets)),
+ Arc::new(entries),
+ nulls,
+ ))
+ }
+
+ #[test]
+ fn blob_row_map_preserves_null_keys_empty_maps_and_parent_nulls() {
+ let input = blob_map_rows(
+ vec![None, Some("named")],
+ vec![Some(b"value"), None],
+ vec![0, 2, 2, 2],
+ Some(NullBuffer::from(vec![true, true, false])),
+ None,
+ );
+ let (field, output) = normalize_blob_write_field(&input,
&blob_map_field(true)).unwrap();
+ output.to_data().validate_full().unwrap();
+ assert_eq!(field.data_type(), output.data_type());
+ let lists = output.as_any().downcast_ref::<ListArray>().unwrap();
+ assert_eq!(lists.value_length(0), 2);
+ assert_eq!(lists.value_length(1), 0);
+ assert!(lists.is_null(2));
+ let entries = lists.value(0);
+ let entries = entries.as_any().downcast_ref::<StructArray>().unwrap();
+ let keys = entries
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let values = entries
+ .column(1)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap();
+ assert!(keys.is_null(0));
+ assert_eq!(keys.value(1), "named");
+ assert_eq!(values.value(0), b"value");
+ assert!(values.is_null(1));
+ }
+
+ #[test]
+ fn blob_map_hidden_entries_do_not_fail_visible_not_null_values() {
+ let input = blob_map_rows(
+ vec![None, Some("visible")],
+ vec![None, Some(b"value")],
+ vec![0, 1, 2],
+ Some(NullBuffer::from(vec![false, true])),
+ Some(NullBuffer::from(vec![false, true])),
+ );
+ let (_, output) = normalize_blob_write_field(&input,
&blob_map_field(false)).unwrap();
+ output.to_data().validate_full().unwrap();
+ let lists = output.as_any().downcast_ref::<ListArray>().unwrap();
+ assert!(lists.is_null(0));
+ assert_eq!(lists.value_length(0), 0);
+ assert_eq!(lists.value_length(1), 1);
+ let entries = lists.value(1);
+ let entries = entries.as_any().downcast_ref::<StructArray>().unwrap();
+ assert_eq!(
+ entries
+ .column(1)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap()
+ .value(0),
+ b"value"
+ );
+ }
+
+ #[test]
+ fn blob_map_visible_null_entries_and_not_null_values_are_rejected() {
+ for entry_nulls in [None, Some(NullBuffer::from(vec![false]))] {
+ let input = blob_map_rows(vec![Some("key")], vec![None], vec![0,
1], None, entry_nulls);
+ assert!(normalize_blob_write_field(&input,
&blob_map_field(false)).is_err());
+ }
+ }
+
+ #[test]
+ fn blob_map_entry_aliases_follow_positional_roles() {
+ let input = blob_map_rows(
+ vec![Some("key")],
+ vec![Some(b"value")],
+ vec![0, 1],
+ None,
+ None,
+ );
+ let expected = blob_map_field(true);
+ let entries =
input.as_any().downcast_ref::<ListArray>().unwrap().value(0);
+ let entries = entries.as_any().downcast_ref::<StructArray>().unwrap();
+ let fields = vec![
+ Field::new("other_key", DataType::Utf8, false),
+ Field::new("other_value", DataType::LargeBinary, true),
+ ]
+ .into();
+ let entries = StructArray::new(fields, entries.columns().to_vec(),
None);
+ let map: ArrayRef = Arc::new(MapArray::new(
+ Arc::new(Field::new(
+ "aliased_entries",
+ entries.data_type().clone(),
+ false,
+ )),
+ OffsetBuffer::new(ScalarBuffer::from(vec![0, 1])),
+ entries,
+ None,
+ false,
+ ));
+ let (_, row_input) = normalize_blob_write_field(&input,
&expected).unwrap();
+ let (_, arrow_input) = normalize_blob_write_field(&map,
&expected).unwrap();
+ assert_eq!(row_input.to_data(), arrow_input.to_data());
+ }
+
+ #[test]
+ fn ordinary_map_writes_cannot_accept_blob_row_transport() {
+ let input = blob_map_rows(vec![None], vec![Some(b"value")], vec![0,
1], None, None);
+ assert!(normalize_write_array(&input,
blob_map_field(true).data_type()).is_err());
+ }
+
+ #[test]
+ fn decimal_blob_map_keys_round_half_up_and_check_precision() {
+ for (text, expected) in [
+ ("12.345", 1235),
+ ("-12.345", -1235),
+ ("12.344", 1234),
+ ("+00012.3", 1230),
+ (".005", 1),
+ ("-.005", -1),
+ ("0.004999999999999999999999999999999999999999999", 0),
+ ("9.999999999999999999999999999999999999999999999", 1000),
+ ] {
+ assert_eq!(decimal_blob_key(text, 10, 2), Some(expected),
"{text}");
+ }
+ for text in [
+ "",
+ ".",
+ "12..3",
+ " 1",
+ "1e2",
+ "99999999.995",
+ "-99999999.995",
+ ] {
+ assert_eq!(decimal_blob_key(text, 10, 2), None, "{text}");
+ }
+ assert_eq!(
+ decimal_blob_key("12345678901234567890.1234567890123456785", 38,
18),
+ Some(12345678901234567890123456789012345679)
+ );
+ }
+
+ #[test]
+ fn decimal_blob_map_transport_preserves_null_and_prunes_hidden_keys() {
+ let input = blob_map_rows(
+ vec![Some("99999999.995"), Some("12.345"), None],
+ vec![Some(b"hidden"), Some(b"rounded"), Some(b"null key")],
+ vec![0, 1, 3],
+ Some(NullBuffer::from(vec![false, true])),
+ None,
+ );
+ let expected = Field::new(
+ "payload",
+ DataType::Map(
+ Arc::new(Field::new(
+ "entries",
+ DataType::Struct(
+ vec![
+ Field::new("key", DataType::Decimal128(10, 2),
false),
+ Field::new("value", DataType::LargeBinary, true),
+ ]
+ .into(),
+ ),
+ false,
+ )),
+ false,
+ ),
+ true,
+ );
+ let (_, output) = normalize_blob_write_field(&input,
&expected).unwrap();
+ let lists = output.as_any().downcast_ref::<ListArray>().unwrap();
+ assert!(lists.is_null(0));
+ let values = lists.value(1);
+ let entries = values.as_any().downcast_ref::<StructArray>().unwrap();
+ let keys = entries
+ .column(0)
+ .as_any()
+ .downcast_ref::<Decimal128Array>()
+ .unwrap();
+ assert_eq!(keys.value(0), 1235);
+ assert!(keys.is_null(1));
+ assert_eq!(keys.data_type(), &DataType::Decimal128(10, 2));
+ let visible_overflow = blob_map_rows(
+ vec![Some("99999999.995")],
+ vec![Some(b"value")],
+ vec![0, 1],
+ None,
+ None,
+ );
+ assert!(normalize_blob_write_field(&visible_overflow,
&expected).is_err());
+ }
+
+ #[test]
+ fn sliced_blob_map_inputs_drop_unreferenced_entries() {
+ let input = blob_map_rows(
+ vec![Some("before"), None, Some("after")],
+ vec![Some(b"before"), Some(b"value"), Some(b"after")],
+ vec![0, 1, 2, 3],
+ None,
+ None,
+ );
+ let (_, output) =
+ normalize_blob_write_field(&input.slice(1, 1),
&blob_map_field(true)).unwrap();
+ let lists = output.as_any().downcast_ref::<ListArray>().unwrap();
+ assert_eq!(lists.offsets().as_ref(), &[0, 1]);
+ let entries = lists.value(0);
+ let entries = entries.as_any().downcast_ref::<StructArray>().unwrap();
+ assert!(entries.column(0).is_null(0));
+ assert_eq!(
+ entries
+ .column(1)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap()
+ .value(0),
+ b"value"
+ );
+ }
+
#[test]
fn fixed_binary_conversion_preserves_nulls_and_bytes() {
let input: ArrayRef = Arc::new(
diff --git a/crates/paimon/tests/blob_update_test.rs
b/crates/paimon/tests/blob_update_test.rs
index a9f5346e..c5a651bc 100644
--- a/crates/paimon/tests/blob_update_test.rs
+++ b/crates/paimon/tests/blob_update_test.rs
@@ -811,6 +811,22 @@ struct BlobSource {
struct BlobSourceFactory(Arc<BlobSource>);
+struct SelectiveBlobSourceFactory(Arc<BlobSource>);
+
+impl paimon::io::UriReaderFactory for SelectiveBlobSourceFactory {
+ fn supports_uri(&self, uri: &str) -> paimon::Result<bool> {
+ Ok(uri.starts_with("custom://"))
+ }
+
+ fn create(&self, uri: &str) -> paimon::Result<Arc<dyn
paimon::io::UriReader>> {
+ assert_eq!(
+ uri, "custom://update",
+ "shadowed source must never be opened"
+ );
+ Ok(self.0.clone())
+ }
+}
+
impl paimon::io::UriReaderFactory for BlobSourceFactory {
fn create(&self, uri: &str) -> paimon::Result<Arc<dyn
paimon::io::UriReader>> {
assert_eq!(uri, "custom://update");
@@ -888,3 +904,259 @@ async fn
row_id_blob_reader_factory_streams_unknown_lengths_and_closes_sources()
assert_eq!(rows[1].1, Some(b"streamed value".to_vec()));
assert_eq!(rows[4].1, Some(b"streamed value".to_vec()));
}
+
+fn source_reader() -> Arc<BlobSource> {
+ Arc::new(BlobSource {
+ opened: AtomicUsize::new(0),
+ closed: Arc::new(AtomicUsize::new(0)),
+ })
+}
+
+fn upsert_rows(ids: Vec<i32>, payloads: Vec<Option<&[u8]>>, values: Vec<i32>)
-> RecordBatch {
+ RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(ids)) as ArrayRef),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from(payloads)) as ArrayRef,
+ ),
+ ("value", Arc::new(Int32Array::from(values)) as ArrayRef),
+ ])
+ .unwrap()
+}
+
+#[tokio::test]
+async fn table_update_passes_reader_factory_to_new_row_id_updaters() {
+ let table = table(&[]).await;
+ seed(&table).await;
+ let reader = source_reader();
+ let builder = table.new_write_builder();
+ let mut update = builder.new_update().unwrap();
+
update.with_blob_uri_reader_factory(Some(Arc::new(BlobSourceFactory(reader.clone()))));
+ let descriptor = BlobDescriptor::new("custom://update".into(), 0,
-1).serialize();
+ let messages = update
+ .update_by_arrow_with_row_id(vec![matched(vec![1],
vec![Some(&descriptor)])])
+ .await
+ .unwrap();
+ assert_eq!(reader.opened.load(Ordering::Relaxed), 1);
+ assert_eq!(reader.closed.load(Ordering::Relaxed), 1);
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(
+ rows(&table).await[1],
+ (1, Some(b"streamed value".to_vec()), 11)
+ );
+}
+
+#[tokio::test]
+async fn
predicate_updates_share_configured_reader_across_scalar_and_array_groups() {
+ use paimon::spec::{Datum, PredicateBuilder};
+ use paimon::table::UpdateAssignment;
+ for array_assignment in [false, true] {
+ let table = table(&[("target-file-row-num", "2")]).await;
+ seed_chunks(&table, 2).await;
+ let before = rows(&table).await;
+ let reader = source_reader();
+ let builder = table.new_write_builder();
+ let mut update = builder.new_update().unwrap();
+
update.with_blob_uri_reader_factory(Some(Arc::new(BlobSourceFactory(reader.clone()))));
+ let reference = BlobDescriptor::new("custom://update".into(), 0,
-1).serialize();
+ let values: ArrayRef = Arc::new(LargeBinaryArray::from(vec![
+ Some(reference.as_slice());
+ if array_assignment {
+ 2
+ } else {
+ 1
+ }
+ ]));
+ let assignment = if array_assignment {
+ UpdateAssignment::Array(vec![values])
+ } else {
+ UpdateAssignment::Scalar(values)
+ };
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .greater_or_equal("id", Datum::Int(3))
+ .unwrap();
+ let messages = update
+ .update_by_predicate(
+ Some(predicate),
+ vec![("payload".into(), assignment)],
+ vec![],
+ )
+ .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 actual = rows(&table).await;
+ assert_eq!(actual[..3], before[..3]);
+ assert_eq!(actual[3], (3, Some(b"streamed value".to_vec()), 13));
+ assert_eq!(actual[4], (4, Some(b"streamed value".to_vec()), 14));
+ }
+}
+
+#[tokio::test]
+async fn
upsert_uses_one_reader_configuration_for_matches_appends_and_native_references()
{
+ let table = table(&[]).await;
+ seed(&table).await;
+ table
+ .file_io()
+ .new_output("memory:/native-source")
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"prefixNATIVEtail"))
+ .await
+ .unwrap();
+ let native = BlobDescriptor::new("memory:/native-source".into(), 6,
6).serialize();
+ let custom = BlobDescriptor::new("custom://update".into(), 0,
-1).serialize();
+ let shadow = BlobDescriptor::new("custom://shadowed".into(), 0,
-1).serialize();
+ let reader = source_reader();
+ let builder = table.new_write_builder();
+ let mut update = builder.new_update().unwrap();
+ update.with_update_type(vec!["payload".into()]).unwrap();
+
update.with_blob_uri_reader_factory(Some(Arc::new(SelectiveBlobSourceFactory(reader.clone()))));
+ let messages = update
+ .upsert_by_arrow_with_key(
+ vec![
+ upsert_rows(
+ vec![1, 8],
+ vec![Some(&shadow), Some(&shadow)],
+ vec![111, 888],
+ ),
+ upsert_rows(
+ vec![1, 8, 9],
+ vec![Some(&custom), Some(&custom), Some(&native)],
+ vec![112, 88, 99],
+ ),
+ ],
+ vec!["id".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 result = rows(&table).await;
+ assert_eq!(result[1], (1, Some(b"streamed value".to_vec()), 11));
+ assert_eq!(result[5], (8, Some(b"streamed value".to_vec()), 88));
+ assert_eq!(result[6], (9, Some(b"NATIVE".to_vec()), 99));
+}
+
+#[tokio::test]
+async fn selective_reader_preserves_native_io_for_ordinary_append_references()
{
+ let table = table(&[]).await;
+ table
+ .file_io()
+ .new_output("memory:/native-source")
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"prefixNATIVEtail"))
+ .await
+ .unwrap();
+ let native = BlobDescriptor::new("memory:/native-source".into(), 6,
6).serialize();
+ let custom = BlobDescriptor::new("custom://update".into(), 0,
-1).serialize();
+ let reader = source_reader();
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+
.with_blob_uri_reader_factory(Some(Arc::new(SelectiveBlobSourceFactory(reader.clone()))))
+ .unwrap();
+ writer
+ .write_arrow_batch(&upsert_rows(
+ vec![0, 1],
+ vec![Some(&native), Some(&custom)],
+ vec![10, 11],
+ ))
+ .await
+ .unwrap();
+ assert_eq!(reader.opened.load(Ordering::Relaxed), 1);
+ assert_eq!(reader.closed.load(Ordering::Relaxed), 1);
+ let messages = writer.prepare_commit().await.unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(
+ rows(&table).await,
+ vec![
+ (0, Some(b"NATIVE".to_vec()), 10),
+ (1, Some(b"streamed value".to_vec()), 11)
+ ]
+ );
+}
+
+#[tokio::test]
+async fn merge_uses_configured_readers_for_matched_and_inserted_blobs() {
+ use paimon::table::{MergeAssignment, MergeSource, WhenMatched,
WhenNotMatched};
+ let table = table(&[]).await;
+ seed(&table).await;
+ let reader = source_reader();
+ let descriptor = BlobDescriptor::new("custom://update".into(), 0,
-1).serialize();
+ let builder = table.new_write_builder();
+ let mut update = builder.new_update().unwrap();
+
update.with_blob_uri_reader_factory(Some(Arc::new(BlobSourceFactory(reader.clone()))));
+ let messages = update
+ .merge_into(
+ MergeSource::Batches(vec![upsert_rows(
+ vec![1, 8],
+ vec![Some(&descriptor), Some(&descriptor)],
+ vec![111, 88],
+ )]),
+ vec![("id".into(), "id".into())],
+ vec![WhenMatched {
+ condition: None,
+ delete: false,
+ assignments: vec![(
+ "payload".into(),
+ MergeAssignment::SourceColumn("payload".into()),
+ )],
+ }],
+ vec![WhenNotMatched {
+ condition: None,
+ assignments: ["id", "payload", "value"]
+ .into_iter()
+ .map(|name| (name.into(),
MergeAssignment::SourceColumn(name.into())))
+ .collect(),
+ }],
+ )
+ .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 result = rows(&table).await;
+ assert_eq!(result[1], (1, Some(b"streamed value".to_vec()), 11));
+ assert_eq!(result[5], (8, Some(b"streamed value".to_vec()), 88));
+}
+
+struct FailedReaderSelection;
+
+impl paimon::io::UriReaderFactory for FailedReaderSelection {
+ fn supports_uri(&self, _uri: &str) -> paimon::Result<bool> {
+ Err(paimon::Error::DataInvalid {
+ message: "reader selection failed".into(),
+ source: None,
+ })
+ }
+
+ fn create(&self, _uri: &str) -> paimon::Result<Arc<dyn
paimon::io::UriReader>> {
+ panic!("Selection failures must not open another reader")
+ }
+}
+
+#[tokio::test]
+async fn
reader_selection_error_does_not_fall_back_to_readable_native_reference() {
+ let table = table(&[]).await;
+ table
+ .file_io()
+ .new_output("memory:/native-source")
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"NATIVE"))
+ .await
+ .unwrap();
+ let descriptor = BlobDescriptor::new("memory:/native-source".into(), 0,
6).serialize();
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ writer
+ .with_blob_uri_reader_factory(Some(Arc::new(FailedReaderSelection)))
+ .unwrap();
+ let error = writer
+ .write_arrow_batch(&upsert_rows(vec![0], vec![Some(&descriptor)],
vec![10]))
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("reader selection failed"));
+ writer.close().await;
+ assert!(physical_files(&table).await.is_empty());
+}
