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 e5b3be5 feat: Implement Blob View support (#486)
e5b3be5 is described below
commit e5b3be5f964f26bc5e85aa937f129ab5b8c28b7a
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 8 23:20:16 2026 +0800
feat: Implement Blob View support (#486)
---
crates/integrations/datafusion/src/blob_view.rs | 267 ++++++++++
crates/integrations/datafusion/src/lib.rs | 2 +
crates/integrations/datafusion/src/sql_context.rs | 36 +-
crates/integrations/datafusion/tests/blob_tests.rs | 164 +++++-
crates/paimon/src/catalog/rest/rest_catalog.rs | 82 +--
crates/paimon/src/spec/blob_view_struct.rs | 217 ++++++++
crates/paimon/src/spec/core_options.rs | 80 ++-
crates/paimon/src/spec/mod.rs | 3 +
crates/paimon/src/table/blob_resolver.rs | 227 ++++++++
crates/paimon/src/table/data_evolution_reader.rs | 585 ++++++++++++++++++---
.../src/table/dedicated_format_file_writer.rs | 110 ++--
crates/paimon/src/table/mod.rs | 1 +
crates/paimon/src/table/rest_env.rs | 118 ++++-
crates/paimon/src/table/table_read.rs | 3 +
crates/paimon/src/table/table_write.rs | 39 +-
crates/paimon/tests/rest_catalog_test.rs | 237 ++++++++-
docs/src/sql.md | 76 ++-
17 files changed, 2015 insertions(+), 232 deletions(-)
diff --git a/crates/integrations/datafusion/src/blob_view.rs
b/crates/integrations/datafusion/src/blob_view.rs
new file mode 100644
index 0000000..345da1f
--- /dev/null
+++ b/crates/integrations/datafusion/src/blob_view.rs
@@ -0,0 +1,267 @@
+// 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.
+
+use std::collections::HashMap;
+use std::hash::{Hash, Hasher};
+use std::sync::Arc;
+
+use datafusion::arrow::array::{
+ Array, BinaryBuilder, Int16Array, Int32Array, Int64Array, Int8Array,
LargeStringArray,
+ StringArray, StringViewArray, UInt16Array, UInt32Array, UInt64Array,
UInt8Array,
+};
+use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, FieldRef};
+use datafusion::common::{DataFusionError, Result as DFResult};
+use datafusion::logical_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDF,
ScalarUDFImpl, Signature,
+ Volatility,
+};
+use datafusion::prelude::SessionContext;
+use paimon::catalog::Catalog;
+use paimon::spec::BlobViewStruct;
+
+use crate::error::to_datafusion_error;
+use crate::runtime::block_on_with_runtime;
+use crate::table_function_args::parse_table_identifier;
+
+const FUNCTION_NAME: &str = "blob_view";
+
+pub fn register_blob_view(ctx: &SessionContext, catalog: Arc<dyn Catalog>,
default_database: &str) {
+ ctx.register_udf(ScalarUDF::from(BlobViewFunc::new(
+ catalog,
+ default_database,
+ )));
+}
+
+#[derive(Clone)]
+struct BlobViewFunc {
+ catalog: Arc<dyn Catalog>,
+ default_database: String,
+ signature: Signature,
+ aliases: Vec<String>,
+}
+
+impl BlobViewFunc {
+ fn new(catalog: Arc<dyn Catalog>, default_database: &str) -> Self {
+ Self {
+ catalog,
+ default_database: default_database.to_string(),
+ signature: Signature::any(3, Volatility::Immutable),
+ aliases: vec!["sys.blob_view".to_string()],
+ }
+ }
+
+ fn field_id(&self, table_name: &str, field_name: &str) -> DFResult<i32> {
+ let identifier = parse_table_identifier(FUNCTION_NAME, table_name,
&self.default_database)?;
+ let catalog = Arc::clone(&self.catalog);
+ let table = block_on_with_runtime(
+ async move { catalog.get_table(&identifier).await },
+ "blob_view: catalog access thread panicked",
+ )
+ .map_err(to_datafusion_error)?;
+
+ let field = table
+ .schema()
+ .fields()
+ .iter()
+ .find(|field| field.name() == field_name)
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "blob_view: cannot find blob field {field_name} in
upstream table {table_name}"
+ ))
+ })?;
+ if !field.data_type().is_blob_type() {
+ return Err(DataFusionError::Plan(format!(
+ "blob_view: field {field_name} in upstream table {table_name}
is not a BLOB field"
+ )));
+ }
+ Ok(field.id())
+ }
+}
+
+impl std::fmt::Debug for BlobViewFunc {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("BlobViewFunc")
+ .field("default_database", &self.default_database)
+ .field("aliases", &self.aliases)
+ .finish()
+ }
+}
+
+impl PartialEq for BlobViewFunc {
+ fn eq(&self, other: &Self) -> bool {
+ self.default_database == other.default_database
+ && Arc::ptr_eq(&self.catalog, &other.catalog)
+ }
+}
+
+impl Eq for BlobViewFunc {}
+
+impl Hash for BlobViewFunc {
+ fn hash<H: Hasher>(&self, state: &mut H) {
+ FUNCTION_NAME.hash(state);
+ self.default_database.hash(state);
+ let ptr = Arc::as_ptr(&self.catalog) as *const () as usize;
+ ptr.hash(state);
+ }
+}
+
+impl ScalarUDFImpl for BlobViewFunc {
+ fn name(&self) -> &str {
+ FUNCTION_NAME
+ }
+
+ fn aliases(&self) -> &[String] {
+ &self.aliases
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[ArrowDataType]) ->
DFResult<ArrowDataType> {
+ Ok(ArrowDataType::Binary)
+ }
+
+ fn return_field_from_args(&self, _args: ReturnFieldArgs) ->
DFResult<FieldRef> {
+ Ok(Arc::new(Field::new(
+ FUNCTION_NAME,
+ ArrowDataType::Binary,
+ true,
+ )))
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) ->
DFResult<ColumnarValue> {
+ if args.args.len() != 3 {
+ return Err(DataFusionError::Plan(
+ "blob_view expects 3 arguments: (table_name, field_name_or_id,
row_id)".to_string(),
+ ));
+ }
+
+ let arrays = ColumnarValue::values_to_arrays(&args.args)?;
+ let table_names = arrays[0].as_ref();
+ let fields = arrays[1].as_ref();
+ let row_ids = arrays[2].as_ref();
+ let mut field_id_cache = HashMap::new();
+ let mut builder = BinaryBuilder::new();
+
+ for row in 0..table_names.len() {
+ let Some(table_name) = string_at(table_names, row)? else {
+ builder.append_null();
+ continue;
+ };
+ let Some(row_id) = int64_at(row_ids, row, "row_id")? else {
+ builder.append_null();
+ continue;
+ };
+
+ let field_id = if is_string_array(fields) {
+ let Some(field_name) = string_at(fields, row)? else {
+ builder.append_null();
+ continue;
+ };
+ let cache_key = (table_name.clone(), field_name.clone());
+ match field_id_cache.get(&cache_key) {
+ Some(field_id) => *field_id,
+ None => {
+ let field_id = self.field_id(&table_name,
&field_name)?;
+ field_id_cache.insert(cache_key, field_id);
+ field_id
+ }
+ }
+ } else {
+ match int64_at(fields, row, "field_id")? {
+ Some(field_id) => i32::try_from(field_id).map_err(|_| {
+ DataFusionError::Plan(format!(
+ "blob_view: field_id {field_id} is outside i32
range"
+ ))
+ })?,
+ None => {
+ builder.append_null();
+ continue;
+ }
+ }
+ };
+
+ let identifier =
+ parse_table_identifier(FUNCTION_NAME, &table_name,
&self.default_database)?;
+ let value = BlobViewStruct::new(identifier, field_id, row_id)
+ .serialize()
+ .map_err(to_datafusion_error)?;
+ builder.append_value(value);
+ }
+
+ Ok(ColumnarValue::Array(Arc::new(builder.finish())))
+ }
+}
+
+fn is_string_array(array: &dyn Array) -> bool {
+ matches!(
+ array.data_type(),
+ ArrowDataType::Utf8 | ArrowDataType::LargeUtf8 |
ArrowDataType::Utf8View
+ )
+}
+
+fn string_at(array: &dyn Array, row: usize) -> DFResult<Option<String>> {
+ if array.is_null(row) {
+ return Ok(None);
+ }
+ if let Some(values) = array.as_any().downcast_ref::<StringArray>() {
+ return Ok(Some(values.value(row).to_string()));
+ }
+ if let Some(values) = array.as_any().downcast_ref::<LargeStringArray>() {
+ return Ok(Some(values.value(row).to_string()));
+ }
+ if let Some(values) = array.as_any().downcast_ref::<StringViewArray>() {
+ return Ok(Some(values.value(row).to_string()));
+ }
+ Err(DataFusionError::Plan(format!(
+ "blob_view: expected string argument, got {:?}",
+ array.data_type()
+ )))
+}
+
+fn int64_at(array: &dyn Array, row: usize, name: &str) ->
DFResult<Option<i64>> {
+ if array.is_null(row) {
+ return Ok(None);
+ }
+ macro_rules! downcast_int {
+ ($ty:ty) => {
+ if let Some(values) = array.as_any().downcast_ref::<$ty>() {
+ return Ok(Some(values.value(row) as i64));
+ }
+ };
+ }
+ downcast_int!(Int8Array);
+ downcast_int!(Int16Array);
+ downcast_int!(Int32Array);
+ downcast_int!(Int64Array);
+ downcast_int!(UInt8Array);
+ downcast_int!(UInt16Array);
+ downcast_int!(UInt32Array);
+ if let Some(values) = array.as_any().downcast_ref::<UInt64Array>() {
+ return i64::try_from(values.value(row)).map(Some).map_err(|_| {
+ DataFusionError::Plan(format!(
+ "blob_view: {name} {} is outside i64 range",
+ values.value(row)
+ ))
+ });
+ }
+ Err(DataFusionError::Plan(format!(
+ "blob_view: expected integer {name}, got {:?}",
+ array.data_type()
+ )))
+}
diff --git a/crates/integrations/datafusion/src/lib.rs
b/crates/integrations/datafusion/src/lib.rs
index c4b79b5..675fda4 100644
--- a/crates/integrations/datafusion/src/lib.rs
+++ b/crates/integrations/datafusion/src/lib.rs
@@ -37,6 +37,7 @@
//! translatable partition-only conjuncts from DataFusion filters.
mod blob_reader;
+mod blob_view;
mod catalog;
mod delete;
mod error;
@@ -69,6 +70,7 @@ use std::sync::{Arc, RwLock};
pub(crate) type DynamicOptions = Arc<RwLock<HashMap<String, String>>>;
pub use blob_reader::BlobReaderRegistry;
+pub use blob_view::register_blob_view;
pub use catalog::{PaimonCatalogProvider, PaimonSchemaProvider};
pub use error::to_datafusion_error;
#[cfg(feature = "fulltext")]
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index babd650..e7b0ba2 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -674,11 +674,13 @@ impl SQLContext {
let identifier = self.resolve_table_name(&ct.name)?;
let mut builder = paimon::spec::Schema::builder();
+ let table_options = extract_options(&ct.table_options)?;
// Columns
for col in &ct.columns {
let paimon_type = column_def_to_paimon_type(col)?;
- builder = builder.column(col.name.value.clone(), paimon_type);
+ let comment = column_def_comment(col);
+ builder = builder.column_with_description(col.name.value.clone(),
paimon_type, comment);
}
// Primary key from constraints: PRIMARY KEY (col, ...)
@@ -707,7 +709,7 @@ impl SQLContext {
}
// Table options from WITH ('key' = 'value', ...)
- for (k, v) in extract_options(&ct.table_options)? {
+ for (k, v) in table_options {
builder = builder.option(k, v);
}
@@ -907,8 +909,7 @@ impl SQLContext {
for op in operations {
match op {
AlterTableOperation::AddColumn { column_def, .. } => {
- let change = column_def_to_add_column(column_def)?;
- changes.push(change);
+ changes.push(column_def_to_add_column(column_def)?);
}
AlterTableOperation::DropColumn {
column_names,
@@ -1706,16 +1707,27 @@ fn extract_partition_by(sql: &str) -> DFResult<(String,
Vec<String>)> {
/// Convert a sqlparser [`ColumnDef`] to a Paimon [`SchemaChange::AddColumn`].
fn column_def_to_add_column(col: &ColumnDef) -> DFResult<SchemaChange> {
let paimon_type = column_def_to_paimon_type(col)?;
- Ok(SchemaChange::add_column(
- col.name.value.clone(),
- paimon_type,
- ))
+ let comment = column_def_comment(col);
+
+ Ok(SchemaChange::AddColumn {
+ field_names: vec![col.name.value.clone()],
+ data_type: paimon_type,
+ comment,
+ column_move: None,
+ })
}
fn column_def_to_paimon_type(col: &ColumnDef) -> DFResult<PaimonDataType> {
sql_data_type_to_paimon_type(&col.data_type, column_def_nullable(col))
}
+fn column_def_comment(col: &ColumnDef) -> Option<String> {
+ col.options.iter().find_map(|opt| match &opt.option {
+ datafusion::sql::sqlparser::ast::ColumnOption::Comment(comment) =>
Some(comment.clone()),
+ _ => None,
+ })
+}
+
fn primary_key_column_name(expr: &SqlExpr) -> String {
match expr {
SqlExpr::Identifier(ident) => ident.value.clone(),
@@ -2579,6 +2591,7 @@ fn register_table_functions(
catalog: &Arc<dyn Catalog>,
default_database: &str,
) {
+ crate::blob_view::register_blob_view(ctx, Arc::clone(catalog),
default_database);
crate::vector_search::register_vector_search(ctx, Arc::clone(catalog),
default_database);
#[cfg(feature = "fulltext")]
crate::full_text_search::register_full_text_search(ctx,
Arc::clone(catalog), default_database);
@@ -2593,7 +2606,7 @@ mod tests {
use async_trait::async_trait;
use paimon::catalog::Database;
- use paimon::spec::{DataType as PaimonDataType, Schema as PaimonSchema};
+ use paimon::spec::{DataType as PaimonDataType, IntType, Schema as
PaimonSchema};
use paimon::table::Table;
// ==================== Mock Catalog ====================
@@ -2620,12 +2633,14 @@ mod tests {
struct MockCatalog {
calls: Mutex<Vec<CatalogCall>>,
+ existing_table: Mutex<Option<Table>>,
}
impl MockCatalog {
fn new() -> Self {
Self {
calls: Mutex::new(Vec::new()),
+ existing_table: Mutex::new(None),
}
}
@@ -2661,6 +2676,9 @@ mod tests {
Ok(())
}
async fn get_table(&self, _identifier: &Identifier) ->
paimon::Result<Table> {
+ if let Some(table) = self.existing_table.lock().unwrap().clone() {
+ return Ok(table);
+ }
Err(paimon::Error::TableNotExist {
full_name: _identifier.to_string(),
})
diff --git a/crates/integrations/datafusion/tests/blob_tests.rs
b/crates/integrations/datafusion/tests/blob_tests.rs
index 3c3d3c6..10d4624 100644
--- a/crates/integrations/datafusion/tests/blob_tests.rs
+++ b/crates/integrations/datafusion/tests/blob_tests.rs
@@ -23,7 +23,7 @@ mod common;
use arrow_array::{Array, BinaryArray, Int32Array, RecordBatch, StringArray};
use common::{create_sql_context, create_test_env, exec};
-use paimon::spec::BlobDescriptor;
+use paimon::spec::{BlobDescriptor, BlobViewStruct};
use paimon_datafusion::SQLContext;
// ======================= Helpers =======================
@@ -706,6 +706,39 @@ async fn
test_blob_descriptor_field_resolve_descriptor_value() {
);
}
+#[tokio::test]
+async fn
test_blob_descriptor_filter_before_resolve_skips_filtered_bad_descriptor() {
+ let (tmp, sql_context) = setup(
+ "CREATE TABLE paimon.test_db.t (\
+ id INT, \
+ name STRING, \
+ picture BLOB \
+ ) WITH (\
+ 'data-evolution.enabled' = 'true', \
+ 'row-tracking.enabled' = 'true', \
+ 'blob-descriptor-field' = 'picture'\
+ )",
+ )
+ .await;
+
+ let missing_uri = format!("file://{}",
tmp.path().join("missing_blob.bin").display());
+ let bad_desc = BlobDescriptor::new(missing_uri, 0, 1);
+ let bad_desc_hex = to_hex(&bad_desc.serialize());
+ let sql = format!(
+ "INSERT INTO paimon.test_db.t (id, name, picture) VALUES \
+ (1, 'Filtered', X'{bad_desc_hex}'), \
+ (2, 'Kept', X'4F4B')"
+ );
+ exec(&sql_context, &sql).await;
+
+ let rows = query_id_name_picture(
+ &sql_context,
+ "SELECT id, name, picture FROM paimon.test_db.t WHERE id = 2",
+ )
+ .await;
+ assert_eq!(rows, vec![(2, "Kept".into(), Some(b"OK".to_vec()))]);
+}
+
/// SET 'paimon.blob-as-descriptor' = 'true' should return serialized
BlobDescriptor
/// bytes instead of the actual blob content.
#[tokio::test]
@@ -765,3 +798,132 @@ async fn test_blob_as_descriptor_dynamic_option() {
assert_eq!(rows[0].2, Some(b"Hello".to_vec()));
assert_eq!(rows[1].2, Some(b"World".to_vec()));
}
+
+#[tokio::test]
+async fn test_blob_view_without_rest_env_preserves_reference() {
+ let (tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog).await;
+ sql_context
+ .sql("CREATE SCHEMA paimon.test_db")
+ .await
+ .unwrap();
+
+ sql_context
+ .sql(
+ "CREATE TABLE paimon.test_db.src (\
+ id INT, \
+ name STRING, \
+ picture BLOB\
+ ) WITH (\
+ 'data-evolution.enabled' = 'true', \
+ 'row-tracking.enabled' = 'true'\
+ )",
+ )
+ .await
+ .unwrap();
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.src (id, name, picture) VALUES \
+ (1, 'Alice', X'616C696365'), \
+ (2, 'Bob', X'626F62')",
+ )
+ .await;
+
+ sql_context
+ .sql(
+ "CREATE TABLE paimon.test_db.view_t (\
+ id INT, \
+ name STRING, \
+ picture BLOB\
+ ) WITH (\
+ 'data-evolution.enabled' = 'true', \
+ 'row-tracking.enabled' = 'true', \
+ 'blob-view-field' = 'picture'\
+ )",
+ )
+ .await
+ .unwrap();
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.view_t (id, name, picture) \
+ SELECT id, name, sys.blob_view('test_db.src', 'picture', \"_ROW_ID\")
\
+ FROM paimon.test_db.src",
+ )
+ .await;
+
+ let rows = query_id_name_picture(
+ &sql_context,
+ "SELECT id, name, picture FROM paimon.test_db.view_t ORDER BY id",
+ )
+ .await;
+ assert_eq!(rows[0].1, "Alice");
+ assert_eq!(rows[1].1, "Bob");
+ let view0_bytes = rows[0].2.as_ref().expect("view bytes should be
preserved");
+ let view1_bytes = rows[1].2.as_ref().expect("view bytes should be
preserved");
+ assert!(BlobViewStruct::is_blob_view_struct(view0_bytes));
+ assert!(BlobViewStruct::is_blob_view_struct(view1_bytes));
+ let view0 = BlobViewStruct::deserialize(view0_bytes).unwrap();
+ let view1 = BlobViewStruct::deserialize(view1_bytes).unwrap();
+ assert_eq!(view0.identifier().full_name(), "test_db.src");
+ assert_eq!(view1.identifier().full_name(), "test_db.src");
+ assert_eq!(view0.row_id(), 0);
+ assert_eq!(view1.row_id(), 1);
+
+ drop(tmp);
+}
+
+#[tokio::test]
+async fn test_blob_view_resolve_disabled_preserves_reference() {
+ let (_tmp, sql_context) = setup(
+ "CREATE TABLE paimon.test_db.src (\
+ id INT, \
+ name STRING, \
+ picture BLOB\
+ ) WITH (\
+ 'data-evolution.enabled' = 'true', \
+ 'row-tracking.enabled' = 'true'\
+ )",
+ )
+ .await;
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.src (id, name, picture) VALUES (1,
'Alice', X'616C696365')",
+ )
+ .await;
+ sql_context
+ .sql(
+ "CREATE TABLE paimon.test_db.view_t (\
+ id INT, \
+ name STRING, \
+ picture BLOB\
+ ) WITH (\
+ 'data-evolution.enabled' = 'true', \
+ 'row-tracking.enabled' = 'true', \
+ 'blob-view-field' = 'picture'\
+ )",
+ )
+ .await
+ .unwrap();
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.view_t (id, name, picture) \
+ SELECT id, name, blob_view('test_db.src', 'picture', \"_ROW_ID\") \
+ FROM paimon.test_db.src",
+ )
+ .await;
+
+ sql_context
+ .sql("SET 'paimon.blob-view.resolve.enabled' = 'false'")
+ .await
+ .unwrap();
+ let rows = query_id_name_picture(
+ &sql_context,
+ "SELECT id, name, picture FROM paimon.test_db.view_t ORDER BY id",
+ )
+ .await;
+ let view_bytes = rows[0].2.as_ref().expect("view bytes should be
preserved");
+ assert!(BlobViewStruct::is_blob_view_struct(view_bytes));
+ let view = BlobViewStruct::deserialize(view_bytes).unwrap();
+ assert_eq!(view.identifier().full_name(), "test_db.src");
+ assert_eq!(view.row_id(), 0);
+}
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index 5ff5d6b..4d87008 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -33,13 +33,10 @@ use crate::catalog::{
};
use crate::common::{CatalogOptions, Options};
use crate::error::Error;
-use crate::io::FileIO;
-use crate::spec::{Partition, Schema, SchemaChange, TableSchema};
+use crate::spec::{Partition, Schema, SchemaChange};
use crate::table::{RESTEnv, Table};
use crate::Result;
-use super::rest_token_file_io::RESTTokenFileIO;
-
/// REST catalog implementation.
///
/// This catalog communicates with a Paimon REST catalog server
@@ -199,76 +196,13 @@ impl Catalog for RESTCatalog {
// ======================= table methods ===============================
async fn get_table(&self, identifier: &Identifier) -> Result<Table> {
- let response = self
- .api
- .get_table(identifier)
- .await
- .map_err(|e| map_rest_error_for_table(e, identifier))?;
-
- // Extract schema from response
- let schema = response.schema.ok_or_else(|| Error::DataInvalid {
- message: format!("Table {} response missing schema",
identifier.full_name()),
- source: None,
- })?;
-
- let schema_id = response.schema_id.ok_or_else(|| Error::DataInvalid {
- message: format!(
- "Table {} response missing schema_id",
- identifier.full_name()
- ),
- source: None,
- })?;
- let table_schema = TableSchema::new(schema_id, &schema);
-
- // Extract table path from response
- let table_path = response.path.ok_or_else(|| Error::DataInvalid {
- message: format!("Table {} response missing path",
identifier.full_name()),
- source: None,
- })?;
-
- // Check if the table is external
- let is_external = response.is_external.ok_or_else(||
Error::DataInvalid {
- message: format!(
- "Table {} response missing is_external",
- identifier.full_name()
- ),
- source: None,
- })?;
-
- // Extract table uuid for RESTEnv
- let uuid = response.id.ok_or_else(|| Error::DataInvalid {
- message: format!(
- "Table {} response missing id (uuid)",
- identifier.full_name()
- ),
- source: None,
- })?;
-
- // Build FileIO based on data_token_enabled and is_external
- // TODO Support token cache and direct oss access
- let file_io = if self.data_token_enabled && !is_external {
- // Use RESTTokenFileIO to get token-based FileIO
- let token_file_io =
- RESTTokenFileIO::new(identifier.clone(), table_path.clone(),
self.options.clone());
- token_file_io.build_file_io().await?
- } else {
- // Mirrors Java RESTCatalog.fileIOFromOptions: build FileIO from
- // catalog options so OSS-backed paths can pick up the
- // user-supplied `fs.oss.*` keys.
- let mut builder = FileIO::from_path(&table_path)?;
- builder = builder.with_props(self.options.to_map());
- builder.build()?
- };
-
- let rest_env = RESTEnv::new(identifier.clone(), uuid,
self.api.clone());
-
- Ok(Table::new(
- file_io,
- identifier.clone(),
- table_path,
- table_schema,
- Some(rest_env),
- ))
+ RESTEnv::load_table(
+ identifier,
+ self.api.clone(),
+ self.options.clone(),
+ self.data_token_enabled,
+ )
+ .await
}
async fn list_tables(&self, database_name: &str) -> Result<Vec<String>> {
diff --git a/crates/paimon/src/spec/blob_view_struct.rs
b/crates/paimon/src/spec/blob_view_struct.rs
new file mode 100644
index 0000000..c76ff69
--- /dev/null
+++ b/crates/paimon/src/spec/blob_view_struct.rs
@@ -0,0 +1,217 @@
+// 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.
+
+use crate::catalog::{Identifier, UNKNOWN_DATABASE};
+use crate::Error;
+
+const CURRENT_VERSION: u8 = 1;
+const MAGIC: u64 = 0x424C4F4256494557; // "BLOBVIEW"
+
+/// Serialized coordinates for a BLOB value stored in an upstream table.
+///
+/// Matches Java `org.apache.paimon.data.BlobViewStruct`: version byte,
+/// `"BLOBVIEW"` magic, UTF-8 `database.table`, field id, and row id, all
+/// little-endian where applicable.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct BlobViewStruct {
+ identifier: Identifier,
+ field_id: i32,
+ row_id: i64,
+}
+
+impl BlobViewStruct {
+ pub fn new(identifier: Identifier, field_id: i32, row_id: i64) -> Self {
+ Self {
+ identifier,
+ field_id,
+ row_id,
+ }
+ }
+
+ pub fn identifier(&self) -> &Identifier {
+ &self.identifier
+ }
+
+ pub fn field_id(&self) -> i32 {
+ self.field_id
+ }
+
+ pub fn row_id(&self) -> i64 {
+ self.row_id
+ }
+
+ pub fn serialize(&self) -> crate::Result<Vec<u8>> {
+ if self.identifier.database() == UNKNOWN_DATABASE {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob view upstream table identifier must include database
name: {}",
+ self.identifier.full_name()
+ ),
+ source: None,
+ });
+ }
+
+ let identifier = self.identifier.full_name();
+ let identifier_bytes = identifier.as_bytes();
+ let identifier_length =
+ i32::try_from(identifier_bytes.len()).map_err(|e|
Error::DataInvalid {
+ message: format!("BlobViewStruct identifier is too long:
{identifier}"),
+ source: Some(Box::new(e)),
+ })?;
+ let total_size = 1 + 8 + 4 + identifier_bytes.len() + 4 + 8;
+ let mut buf = Vec::with_capacity(total_size);
+
+ buf.push(CURRENT_VERSION);
+ buf.extend_from_slice(&MAGIC.to_le_bytes());
+ buf.extend_from_slice(&identifier_length.to_le_bytes());
+ buf.extend_from_slice(identifier_bytes);
+ buf.extend_from_slice(&self.field_id.to_le_bytes());
+ buf.extend_from_slice(&self.row_id.to_le_bytes());
+ Ok(buf)
+ }
+
+ pub fn deserialize(bytes: &[u8]) -> crate::Result<Self> {
+ if bytes.is_empty() {
+ return Err(invalid_payload("too short"));
+ }
+
+ let version = bytes[0];
+ if version != CURRENT_VERSION {
+ return Err(Error::Unsupported {
+ message: format!(
+ "Expecting BlobViewStruct version to be {CURRENT_VERSION},
but found {version}."
+ ),
+ });
+ }
+
+ let mut pos = 1;
+ if bytes.len() < pos + 8 {
+ return Err(invalid_payload("too short"));
+ }
+ let magic = u64::from_le_bytes(bytes[pos..pos +
8].try_into().unwrap());
+ if magic != MAGIC {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Invalid BlobViewStruct: missing magic header. Expected
magic: {MAGIC}, but found: {magic}"
+ ),
+ source: None,
+ });
+ }
+ pos += 8;
+
+ if bytes.len() < pos + 4 {
+ return Err(invalid_payload("too short"));
+ }
+ let identifier_length = i32::from_le_bytes(bytes[pos..pos +
4].try_into().unwrap());
+ if identifier_length < 0 {
+ return Err(invalid_payload(&format!(
+ "negative identifier length: {identifier_length}"
+ )));
+ }
+ let identifier_length = identifier_length as usize;
+ pos += 4;
+
+ if bytes.len() < pos + identifier_length + 4 + 8 {
+ return Err(invalid_payload("identifier length exceeds data size"));
+ }
+ let identifier_text = String::from_utf8(bytes[pos..pos +
identifier_length].to_vec())
+ .map_err(|e| Error::DataInvalid {
+ message: format!("Invalid UTF-8 in BlobViewStruct identifier:
{e}"),
+ source: Some(Box::new(e)),
+ })?;
+ pos += identifier_length;
+
+ let field_id = i32::from_le_bytes(bytes[pos..pos +
4].try_into().unwrap());
+ pos += 4;
+ let row_id = i64::from_le_bytes(bytes[pos..pos +
8].try_into().unwrap());
+ pos += 8;
+
+ if pos != bytes.len() {
+ return Err(invalid_payload("trailing bytes"));
+ }
+
+ Ok(Self {
+ identifier: parse_identifier(&identifier_text)?,
+ field_id,
+ row_id,
+ })
+ }
+
+ pub fn is_blob_view_struct(bytes: &[u8]) -> bool {
+ if bytes.len() < 9 {
+ return false;
+ }
+ let version = bytes[0];
+ if version != CURRENT_VERSION {
+ return false;
+ }
+ let magic = u64::from_le_bytes(bytes[1..9].try_into().unwrap());
+ magic == MAGIC
+ }
+}
+
+fn parse_identifier(full_name: &str) -> crate::Result<Identifier> {
+ let Some((database, object)) = full_name.rsplit_once('.') else {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "BlobViewStruct identifier must be 'database.table', got:
{full_name}"
+ ),
+ source: None,
+ });
+ };
+ Ok(Identifier::new(database, object))
+}
+
+fn invalid_payload(message: &str) -> Error {
+ Error::DataInvalid {
+ message: format!("Invalid BlobViewStruct data: {message}"),
+ source: None,
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_serialize_deserialize_roundtrip() {
+ let view = BlobViewStruct::new(Identifier::new("db", "source"), 3, 42);
+ let bytes = view.serialize().unwrap();
+
+ assert!(BlobViewStruct::is_blob_view_struct(&bytes));
+ assert_eq!(BlobViewStruct::deserialize(&bytes).unwrap(), view);
+ }
+
+ #[test]
+ fn test_rejects_unknown_database_on_serialize() {
+ let view = BlobViewStruct::new(Identifier::new(UNKNOWN_DATABASE,
"source"), 3, 42);
+ assert!(view.serialize().is_err());
+ }
+
+ #[test]
+ fn test_rejects_trailing_bytes() {
+ let mut bytes = BlobViewStruct::new(Identifier::new("db", "source"),
3, 42)
+ .serialize()
+ .unwrap();
+ bytes.push(1);
+
+ let err = BlobViewStruct::deserialize(&bytes).unwrap_err();
+ assert!(
+ matches!(err, Error::DataInvalid { message, .. } if
message.contains("trailing bytes"))
+ );
+ }
+}
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index 8585e37..a28b0fa 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -94,6 +94,8 @@ const DEFAULT_GLOBAL_INDEX_ROW_COUNT_PER_SHARD: i64 = 100_000;
const DEFAULT_GLOBAL_INDEX_FALLBACK_SCAN_MAX_SIZE: i64 = 256 * 1024 * 1024;
const BLOB_AS_DESCRIPTOR_OPTION: &str = "blob-as-descriptor";
const BLOB_DESCRIPTOR_FIELD_OPTION: &str = "blob-descriptor-field";
+const BLOB_VIEW_FIELD_OPTION: &str = "blob-view-field";
+pub const BLOB_VIEW_RESOLVE_ENABLED_OPTION: &str = "blob-view.resolve.enabled";
/// Merge engine for primary-key tables.
///
@@ -834,9 +836,43 @@ impl<'a> CoreOptions<'a> {
/// Comma-separated BLOB field names stored as serialized BlobDescriptor
/// bytes inline in normal data files (no .blob files for these fields).
pub fn blob_descriptor_fields(&self) -> HashSet<String> {
+ self.parse_csv_set(BLOB_DESCRIPTOR_FIELD_OPTION)
+ }
+
+ /// Comma-separated BLOB field names stored as serialized BlobViewStruct
+ /// bytes inline in normal data files.
+ pub fn blob_view_fields(&self) -> HashSet<String> {
+ self.parse_csv_set(BLOB_VIEW_FIELD_OPTION)
+ }
+
+ /// Whether blob-view fields should resolve upstream BLOB values on reads.
+ /// Default is true, matching Java `CoreOptions.BLOB_VIEW_RESOLVE_ENABLED`.
+ pub fn blob_view_resolve_enabled(&self) -> bool {
+ self.options
+ .get(BLOB_VIEW_RESOLVE_ENABLED_OPTION)
+ .map(|v| v.eq_ignore_ascii_case("true"))
+ .unwrap_or(true)
+ }
+
+ /// BLOB fields stored inline in normal data files: descriptor fields plus
+ /// view fields. Non-inline BLOB fields are written to dedicated `.blob`
+ /// files.
+ pub fn blob_inline_fields(&self) -> HashSet<String> {
+ let mut fields = self.blob_descriptor_fields();
+ fields.extend(self.blob_view_fields());
+ fields
+ }
+
+ fn parse_csv_set(&self, option_name: &'static str) -> HashSet<String> {
self.options
- .get(BLOB_DESCRIPTOR_FIELD_OPTION)
- .map(|s| s.split(',').map(|f| f.trim().to_string()).collect())
+ .get(option_name)
+ .map(|s| {
+ s.split(',')
+ .map(str::trim)
+ .filter(|f| !f.is_empty())
+ .map(ToString::to_string)
+ .collect()
+ })
.unwrap_or_default()
}
}
@@ -1409,6 +1445,46 @@ mod tests {
assert_eq!(core.write_parquet_buffer_size(), 32 * 1024 * 1024);
}
+ #[test]
+ fn test_blob_view_options() {
+ let options = HashMap::from([
+ (
+ BLOB_DESCRIPTOR_FIELD_OPTION.to_string(),
+ "thumb, payload".to_string(),
+ ),
+ (
+ BLOB_VIEW_FIELD_OPTION.to_string(),
+ "image, video".to_string(),
+ ),
+ ]);
+ let core = CoreOptions::new(&options);
+
+ assert_eq!(
+ core.blob_descriptor_fields(),
+ HashSet::from(["thumb".to_string(), "payload".to_string()])
+ );
+ assert_eq!(
+ core.blob_view_fields(),
+ HashSet::from(["image".to_string(), "video".to_string()])
+ );
+ assert_eq!(
+ core.blob_inline_fields(),
+ HashSet::from([
+ "thumb".to_string(),
+ "payload".to_string(),
+ "image".to_string(),
+ "video".to_string()
+ ])
+ );
+ assert!(core.blob_view_resolve_enabled());
+
+ let disabled = HashMap::from([(
+ BLOB_VIEW_RESOLVE_ENABLED_OPTION.to_string(),
+ "false".to_string(),
+ )]);
+ assert!(!CoreOptions::new(&disabled).blob_view_resolve_enabled());
+ }
+
#[test]
fn test_validate_scan_options_rejects_unsupported() {
for key in [
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index eb4dc54..9e8b07b 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -25,6 +25,9 @@ pub use binary_row::*;
mod blob_descriptor;
pub use blob_descriptor::BlobDescriptor;
+mod blob_view_struct;
+pub use blob_view_struct::BlobViewStruct;
+
mod data_file;
pub use data_file::*;
diff --git a/crates/paimon/src/table/blob_resolver.rs
b/crates/paimon/src/table/blob_resolver.rs
new file mode 100644
index 0000000..bbbf24b
--- /dev/null
+++ b/crates/paimon/src/table/blob_resolver.rs
@@ -0,0 +1,227 @@
+// 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.
+
+use crate::io::{FileIO, FileRead};
+use crate::spec::BlobDescriptor;
+use crate::Result;
+use arrow_array::builder::BinaryBuilder;
+use arrow_array::{Array, BinaryArray};
+use bytes::Bytes;
+use std::collections::HashMap;
+
+const BLOB_RANGE_MERGE_GAP: u64 = 64 * 1024;
+const BLOB_RANGE_MERGE_MAX_SPAN: u64 = 8 * 1024 * 1024;
+
+/// For each row in a blob column, if the value is a serialized
`BlobDescriptor`,
+/// resolve it by reading the actual data from the referenced
URI+offset+length.
+/// Raw data values are passed through unchanged.
+pub(crate) async fn resolve_blob_column(
+ col: &BinaryArray,
+ file_io: &FileIO,
+) -> Result<BinaryArray> {
+ let mut needs_resolve = false;
+ for i in 0..col.len() {
+ if !col.is_null(i) && BlobDescriptor::is_blob_descriptor(col.value(i))
{
+ needs_resolve = true;
+ break;
+ }
+ }
+
+ if !needs_resolve {
+ return Ok(col.clone());
+ }
+
+ let mut cells = Vec::with_capacity(col.len());
+ let mut requests_by_uri: HashMap<String, Vec<BlobReadRequest>> =
HashMap::new();
+ let mut value_capacity = 0usize;
+
+ for row in 0..col.len() {
+ if col.is_null(row) {
+ cells.push(ResolvedBlobCell::Null);
+ continue;
+ }
+
+ let value = col.value(row);
+ if BlobDescriptor::is_blob_descriptor(value) {
+ let desc = BlobDescriptor::deserialize(value)?;
+ let offset = u64::try_from(desc.offset()).map_err(|e|
crate::Error::DataInvalid {
+ message: format!(
+ "BlobDescriptor offset must be non-negative: {}",
+ desc.offset()
+ ),
+ source: Some(Box::new(e)),
+ })?;
+ let length = u64::try_from(desc.length()).map_err(|e|
crate::Error::DataInvalid {
+ message: format!(
+ "BlobDescriptor length must be non-negative: {}",
+ desc.length()
+ ),
+ source: Some(Box::new(e)),
+ })?;
+ offset
+ .checked_add(length)
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!(
+ "BlobDescriptor range overflows u64: offset={offset},
length={length}"
+ ),
+ source: None,
+ })?;
+ value_capacity = value_capacity.saturating_add(length as usize);
+ requests_by_uri
+ .entry(desc.uri().to_string())
+ .or_default()
+ .push(BlobReadRequest {
+ row,
+ offset,
+ length,
+ });
+ cells.push(ResolvedBlobCell::Null);
+ } else {
+ value_capacity = value_capacity.saturating_add(value.len());
+ cells.push(ResolvedBlobCell::Value(Bytes::copy_from_slice(value)));
+ }
+ }
+
+ let mut readers: HashMap<String, Box<dyn FileRead>> = HashMap::new();
+ for (uri, requests) in requests_by_uri {
+ if !readers.contains_key(&uri) {
+ let input = file_io.new_input(&uri)?;
+ let reader = input.reader().await?;
+ readers.insert(uri.clone(), Box::new(reader));
+ }
+ let reader = readers.get(&uri).unwrap();
+ for merged in merge_blob_read_requests(requests) {
+ let data = reader.read(merged.start..merged.end).await?;
+ for request in merged.requests {
+ let start = (request.offset - merged.start) as usize;
+ let end = start + request.length as usize;
+ cells[request.row] =
ResolvedBlobCell::Value(data.slice(start..end));
+ }
+ }
+ }
+
+ let mut builder = BinaryBuilder::with_capacity(col.len(), value_capacity);
+ for cell in cells {
+ match cell {
+ ResolvedBlobCell::Null => builder.append_null(),
+ ResolvedBlobCell::Value(value) =>
builder.append_value(value.as_ref()),
+ }
+ }
+ Ok(builder.finish())
+}
+
+#[derive(Debug)]
+enum ResolvedBlobCell {
+ Null,
+ Value(Bytes),
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct BlobReadRequest {
+ row: usize,
+ offset: u64,
+ length: u64,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+struct MergedBlobRead {
+ start: u64,
+ end: u64,
+ requests: Vec<BlobReadRequest>,
+}
+
+fn merge_blob_read_requests(mut requests: Vec<BlobReadRequest>) ->
Vec<MergedBlobRead> {
+ if requests.is_empty() {
+ return Vec::new();
+ }
+
+ requests.sort_by_key(|request| (request.offset, request.length,
request.row));
+ let mut merged = Vec::new();
+ let mut current = MergedBlobRead {
+ start: requests[0].offset,
+ end: requests[0].offset + requests[0].length,
+ requests: vec![requests[0].clone()],
+ };
+
+ for request in requests.into_iter().skip(1) {
+ let request_end = request.offset + request.length;
+ let close_enough = current
+ .end
+ .checked_add(BLOB_RANGE_MERGE_GAP)
+ .is_some_and(|merge_limit| request.offset <= merge_limit);
+ let merged_end = current.end.max(request_end);
+ let merged_span = merged_end - current.start;
+ if close_enough && merged_span <= BLOB_RANGE_MERGE_MAX_SPAN {
+ current.end = merged_end;
+ current.requests.push(request);
+ } else {
+ merged.push(current);
+ current = MergedBlobRead {
+ start: request.offset,
+ end: request_end,
+ requests: vec![request],
+ };
+ }
+ }
+ merged.push(current);
+ merged
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_merge_blob_read_requests_merges_nearby_ranges() {
+ let merged = merge_blob_read_requests(vec![
+ BlobReadRequest {
+ row: 2,
+ offset: 120,
+ length: 5,
+ },
+ BlobReadRequest {
+ row: 0,
+ offset: 0,
+ length: 10,
+ },
+ BlobReadRequest {
+ row: 1,
+ offset: 10,
+ length: 8,
+ },
+ BlobReadRequest {
+ row: 3,
+ offset: BLOB_RANGE_MERGE_MAX_SPAN + 1,
+ length: 4,
+ },
+ ]);
+
+ assert_eq!(merged.len(), 2);
+ assert_eq!(merged[0].start, 0);
+ assert_eq!(merged[0].end, 125);
+ assert_eq!(
+ merged[0]
+ .requests
+ .iter()
+ .map(|request| request.row)
+ .collect::<Vec<_>>(),
+ vec![0, 1, 2]
+ );
+ assert_eq!(merged[1].start, BLOB_RANGE_MERGE_MAX_SPAN + 1);
+ assert_eq!(merged[1].end, BLOB_RANGE_MERGE_MAX_SPAN + 5);
+ }
+}
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index ccb9cae..1c5e9a8 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -23,13 +23,15 @@ use crate::arrow::build_target_arrow_schema;
use crate::arrow::format::FilePredicates;
use crate::deletion_vector::{DeletionVector, DeletionVectorFactory};
use crate::io::FileIO;
-use crate::spec::{DataField, DataFileMeta, DataType, Predicate,
ROW_ID_FIELD_NAME};
+use crate::spec::{
+ BigIntType, BlobDescriptor, BlobViewStruct, DataField, DataFileMeta,
DataType, Predicate,
+ ROW_ID_FIELD_ID, ROW_ID_FIELD_NAME,
+};
use crate::table::dedicated_format_file_writer::is_blob_file_name;
use crate::table::schema_manager::SchemaManager;
-use crate::table::ArrowRecordBatchStream;
-use crate::table::RowRange;
+use crate::table::{ArrowRecordBatchStream, RESTEnv, RowRange};
use crate::{DataSplit, Error};
-use arrow_array::{Array, Int64Array, RecordBatch};
+use arrow_array::{Array, BinaryArray, Int64Array, RecordBatch};
use async_stream::try_stream;
use futures::StreamExt;
use roaring::RoaringBitmap;
@@ -101,6 +103,9 @@ pub(crate) struct DataEvolutionReader {
predicates: Vec<Predicate>,
blob_as_descriptor: bool,
blob_descriptor_fields: HashSet<String>,
+ blob_view_fields: HashSet<String>,
+ blob_view_resolve_enabled: bool,
+ blob_view_rest_env: Option<RESTEnv>,
}
impl DataEvolutionReader {
@@ -114,6 +119,9 @@ impl DataEvolutionReader {
predicates: Vec<Predicate>,
blob_as_descriptor: bool,
blob_descriptor_fields: HashSet<String>,
+ blob_view_fields: HashSet<String>,
+ blob_view_resolve_enabled: bool,
+ blob_view_rest_env: Option<RESTEnv>,
) -> crate::Result<Self> {
let row_id_index = read_type.iter().position(|f| f.name() ==
ROW_ID_FIELD_NAME);
let file_read_type: Vec<DataField> = read_type
@@ -125,7 +133,7 @@ impl DataEvolutionReader {
// Widen the file read set with predicate columns not already projected
// so the residual filter can evaluate every leaf. Extras land strictly
- // at the END: `filter_and_project_output` relies on that to project
the
+ // at the END: `project_output` relies on that to project the
// final batch back to `read_type` by prefix. Predicate leaf indices
// point into the table schema, so `file_fields` = `table_fields`.
let file_predicates = (!predicates.is_empty()).then(|| FilePredicates {
@@ -154,6 +162,9 @@ impl DataEvolutionReader {
predicates,
blob_as_descriptor,
blob_descriptor_fields,
+ blob_view_fields,
+ blob_view_resolve_enabled,
+ blob_view_rest_env,
})
}
@@ -162,6 +173,17 @@ impl DataEvolutionReader {
let splits: Vec<DataSplit> = data_splits.to_vec();
Ok(try_stream! {
+ let resolve_blob_views = !self.blob_view_read_fields().is_empty();
+ let descriptor_fields =
self.descriptor_fields_to_resolve(resolve_blob_views);
+ let filter_before_blob_resolution =
+ self.can_filter_before_blob_resolution(resolve_blob_views,
&descriptor_fields);
+ let blob_view_lookup = self
+ .preload_blob_view_lookup(&splits,
filter_before_blob_resolution)
+ .await?;
+ let descriptor_fields =
self.descriptor_fields_to_resolve(blob_view_lookup.is_some());
+ let filter_before_blob_resolution =
+
self.can_filter_before_blob_resolution(blob_view_lookup.is_some(),
&descriptor_fields);
+
let file_reader = DataFileReader::new(
self.file_io.clone(),
self.schema_manager.clone(),
@@ -224,32 +246,27 @@ impl DataEvolutionReader {
)?;
while let Some(batch) = stream.next().await {
let batch = batch?;
- let batch = if !self.blob_as_descriptor &&
!self.blob_descriptor_fields.is_empty() {
- resolve_descriptor_columns(batch,
&self.blob_descriptor_fields, &self.file_io).await?
- } else {
- batch
- };
let num_rows = batch.num_rows();
- if let Some(idx) = self.row_id_index {
+ let batch = if let Some(idx) = self.row_id_index {
if !has_row_id {
- yield self.filter_and_project_output(
- append_null_row_id_column(batch, idx,
&self.wide_output_schema)?,
- )?;
+ append_null_row_id_column(batch, idx,
&self.wide_output_schema)?
} else if let Some(ref ids) = selected_row_ids
{
- yield self.filter_and_project_output(
- attach_row_id(batch, idx, ids, &mut
row_id_offset, &self.wide_output_schema)?,
- )?;
+ attach_row_id(batch, idx, ids, &mut
row_id_offset, &self.wide_output_schema)?
} else {
let row_ids: Vec<i64> =
(row_id_cursor..row_id_cursor + num_rows as i64).collect();
row_id_cursor += num_rows as i64;
let array: Arc<dyn arrow_array::Array> =
Arc::new(Int64Array::from(row_ids));
- yield self.filter_and_project_output(
- insert_column_at(batch, array, idx,
&self.wide_output_schema)?,
- )?;
+ insert_column_at(batch, array, idx,
&self.wide_output_schema)?
}
} else {
- yield self.filter_and_project_output(batch)?;
- }
+ batch
+ };
+ yield self.finish_wide_batch(
+ batch,
+ blob_view_lookup.as_ref(),
+ &descriptor_fields,
+ filter_before_blob_resolution,
+ ).await?;
}
}
} else {
@@ -298,22 +315,24 @@ impl DataEvolutionReader {
while let Some(batch) = merge_stream.next().await {
let batch = batch?;
let num_rows = batch.num_rows();
- if let Some(idx) = self.row_id_index {
+ let batch = if let Some(idx) = self.row_id_index {
if let Some(ref ids) = selected_row_ids {
- yield self.filter_and_project_output(
- attach_row_id(batch, idx, ids, &mut
row_id_offset, &self.wide_output_schema)?,
- )?;
+ attach_row_id(batch, idx, ids, &mut
row_id_offset, &self.wide_output_schema)?
} else {
let row_ids: Vec<i64> =
(row_id_cursor..row_id_cursor + num_rows as i64).collect();
row_id_cursor += num_rows as i64;
let array: Arc<dyn arrow_array::Array> =
Arc::new(Int64Array::from(row_ids));
- yield self.filter_and_project_output(
- insert_column_at(batch, array, idx,
&self.wide_output_schema)?,
- )?;
+ insert_column_at(batch, array, idx,
&self.wide_output_schema)?
}
} else {
- yield self.filter_and_project_output(batch)?;
- }
+ batch
+ };
+ yield self.finish_wide_batch(
+ batch,
+ blob_view_lookup.as_ref(),
+ &descriptor_fields,
+ filter_before_blob_resolution,
+ ).await?;
}
}
}
@@ -321,40 +340,39 @@ impl DataEvolutionReader {
.boxed())
}
- /// Apply the residual predicates to a wide batch (post `_ROW_ID` attach)
- /// and project it back to the caller's read_type.
+ /// Apply the residual predicates to a wide batch (post `_ROW_ID` attach).
///
- /// Layout invariant: the first `output_schema.fields().len()` columns of
- /// `batch` are exactly the original read_type columns — extras were
- /// appended at the end by `widen_scan_fields`, and `_ROW_ID` insertion at
- /// `row_id_index` keeps them trailing. `_ROW_ID` correctness: ids are
- /// attached before this filter, so surviving rows keep their original ids.
- ///
- /// No predicates implies no widening (extras only come from predicates),
- /// so the empty-predicate fast path returns the batch untouched.
- fn filter_and_project_output(&self, batch: RecordBatch) ->
crate::Result<RecordBatch> {
- let filtered =
- if self.predicates.is_empty() {
- batch
- } else {
- let mask = crate::arrow::residual::evaluate_predicates_mask(
- &batch,
- &self.predicates,
- &self.table_fields,
- &self.wide_file_read_type,
- )?;
- match mask {
- Some(mask) =>
arrow_select::filter::filter_record_batch(&batch, &mask)
- .map_err(|e| Error::DataInvalid {
- message: format!(
- "Failed to filter data-evolution batch by
predicates: {e}"
- ),
- source: Some(Box::new(e)),
- })?,
- None => batch,
+ /// `_ROW_ID` correctness: ids are attached before this filter, so
surviving
+ /// rows keep their original ids.
+ fn filter_wide_batch(&self, batch: RecordBatch) ->
crate::Result<RecordBatch> {
+ if self.predicates.is_empty() {
+ return Ok(batch);
+ }
+
+ let mask = crate::arrow::residual::evaluate_predicates_mask(
+ &batch,
+ &self.predicates,
+ &self.table_fields,
+ &self.wide_file_read_type,
+ )?;
+ match mask {
+ Some(mask) => arrow_select::filter::filter_record_batch(&batch,
&mask).map_err(|e| {
+ Error::DataInvalid {
+ message: format!("Failed to filter data-evolution batch by
predicates: {e}"),
+ source: Some(Box::new(e)),
}
- };
+ }),
+ None => Ok(batch),
+ }
+ }
+ /// Project a wide batch back to the caller's read_type.
+ ///
+ /// Layout invariant: the first `output_schema.fields().len()` columns of
+ /// `batch` are exactly the original read_type columns. Extras were
appended
+ /// at the end by `widen_scan_fields`, and `_ROW_ID` insertion at
+ /// `row_id_index` keeps them trailing.
+ fn project_output(&self, filtered: RecordBatch) ->
crate::Result<RecordBatch> {
let final_width = self.output_schema.fields().len();
if filtered.num_columns() == final_width {
return Ok(filtered);
@@ -375,6 +393,126 @@ impl DataEvolutionReader {
})
}
+ async fn finish_wide_batch(
+ &self,
+ batch: RecordBatch,
+ blob_view_lookup: Option<&BlobViewLookup>,
+ descriptor_fields: &HashSet<String>,
+ filter_before_blob_resolution: bool,
+ ) -> crate::Result<RecordBatch> {
+ let mut batch = if filter_before_blob_resolution {
+ self.filter_wide_batch(batch)?
+ } else {
+ batch
+ };
+ if filter_before_blob_resolution && batch.num_rows() == 0 {
+ return self.project_output(batch);
+ }
+
+ batch = self.resolve_blob_view_columns(batch, blob_view_lookup)?;
+ let mut batch = if !self.blob_as_descriptor &&
!descriptor_fields.is_empty() {
+ resolve_descriptor_columns(batch, descriptor_fields,
&self.file_io).await?
+ } else {
+ batch
+ };
+
+ if !filter_before_blob_resolution {
+ batch = self.filter_wide_batch(batch)?;
+ }
+ self.project_output(batch)
+ }
+
+ fn can_filter_before_blob_resolution(
+ &self,
+ resolve_blob_views: bool,
+ descriptor_fields: &HashSet<String>,
+ ) -> bool {
+ let mut transformed_fields = HashSet::new();
+ if resolve_blob_views {
+ transformed_fields.extend(self.blob_view_fields.iter().cloned());
+ }
+ if !self.blob_as_descriptor {
+ transformed_fields.extend(descriptor_fields.iter().cloned());
+ }
+ transformed_fields.is_empty()
+ || !predicates_reference_any_field(
+ &self.predicates,
+ &transformed_fields,
+ &self.table_fields,
+ )
+ }
+
+ fn blob_view_read_fields(&self) -> Vec<DataField> {
+ if !self.blob_view_resolve_enabled ||
self.blob_view_rest_env.is_none() {
+ return Vec::new();
+ }
+ self.wide_file_read_type
+ .iter()
+ .filter(|field| self.blob_view_fields.contains(field.name()))
+ .cloned()
+ .collect()
+ }
+
+ async fn preload_blob_view_lookup(
+ &self,
+ splits: &[DataSplit],
+ filter_before_blob_resolution: bool,
+ ) -> crate::Result<Option<BlobViewLookup>> {
+ let view_fields = self.blob_view_read_fields();
+ if view_fields.is_empty() {
+ return Ok(None);
+ }
+ let Some(rest_env) = self.blob_view_rest_env.clone() else {
+ return Ok(None);
+ };
+
+ let prescan = DataEvolutionReader::new(
+ self.file_io.clone(),
+ self.schema_manager.clone(),
+ self.table_schema_id,
+ self.table_fields.clone(),
+ view_fields.clone(),
+ if filter_before_blob_resolution {
+ self.predicates.clone()
+ } else {
+ Vec::new()
+ },
+ true,
+ HashSet::new(),
+ HashSet::new(),
+ false,
+ None,
+ )?;
+ let mut stream = prescan.read(splits)?;
+ let mut view_structs = HashSet::new();
+ while let Some(batch) = stream.next().await {
+ let batch = batch?;
+ collect_blob_view_structs(&batch, &self.blob_view_fields, &mut
view_structs)?;
+ }
+
+ BlobViewLookup::load(rest_env, view_structs).await.map(Some)
+ }
+
+ fn descriptor_fields_to_resolve(&self, resolve_blob_views: bool) ->
HashSet<String> {
+ let mut fields = self.blob_descriptor_fields.clone();
+ if resolve_blob_views {
+ fields.extend(self.blob_view_fields.iter().cloned());
+ }
+ fields
+ }
+
+ fn resolve_blob_view_columns(
+ &self,
+ batch: RecordBatch,
+ lookup: Option<&BlobViewLookup>,
+ ) -> crate::Result<RecordBatch> {
+ let Some(lookup) = lookup else {
+ return Ok(batch);
+ };
+
+ replace_blob_view_columns(batch, &self.blob_view_fields, lookup)
+ }
+
/// Merge multiple logical sources column-wise for data evolution.
///
/// Normal partial-column files remain one source per file. Rolling `.blob`
@@ -398,13 +536,10 @@ impl DataEvolutionReader {
let split = split.clone();
let prepared_group = prepared_group.clone();
// The merge plan reads the WIDE field set so widened predicate columns
- // are materialized for the residual filter. Blob-descriptor note: if a
- // widened predicate column is a blob-descriptor field,
- // `resolve_descriptor_columns` resolves it by name in the wide batch
- // before the residual runs, so with `blob_as_descriptor = false` the
- // residual compares resolved bytes; with `true` it compares descriptor
- // bytes. Resolution before filtering can do IO for rows the filter
then
- // drops — accepted trade-off.
+ // are materialized for the residual filter. Blob descriptor/view
+ // resolution is intentionally deferred to `finish_wide_batch`, where
+ // the reader can filter first when predicates do not depend on fields
+ // whose values would be transformed by resolution.
let read_type = self.wide_file_read_type.clone();
let table_fields = self.table_fields.clone();
let blob_descriptor_fields = self.blob_descriptor_fields.clone();
@@ -580,11 +715,6 @@ impl DataEvolutionReader {
source: Some(Box::new(e)),
}
})?;
- let merged = if !blob_as_descriptor &&
!blob_descriptor_fields.is_empty() {
- resolve_descriptor_columns(merged,
&blob_descriptor_fields, &file_io).await?
- } else {
- merged
- };
yield merged;
}
}
@@ -608,9 +738,7 @@ async fn resolve_descriptor_columns(
.as_any()
.downcast_ref::<arrow_array::BinaryArray>()
{
- let resolved =
-
super::dedicated_format_file_writer::resolve_blob_column(bin_col, file_io)
- .await?;
+ let resolved =
super::blob_resolver::resolve_blob_column(bin_col, file_io).await?;
columns.push(Arc::new(resolved));
changed = true;
continue;
@@ -629,6 +757,309 @@ async fn resolve_descriptor_columns(
})
}
+fn predicates_reference_any_field(
+ predicates: &[Predicate],
+ field_names: &HashSet<String>,
+ table_fields: &[DataField],
+) -> bool {
+ predicates
+ .iter()
+ .any(|predicate| predicate_references_any_field(predicate,
field_names, table_fields))
+}
+
+fn predicate_references_any_field(
+ predicate: &Predicate,
+ field_names: &HashSet<String>,
+ table_fields: &[DataField],
+) -> bool {
+ match predicate {
+ Predicate::Leaf { column, index, .. } => {
+ field_names.contains(column)
+ || table_fields
+ .get(*index)
+ .is_some_and(|field| field_names.contains(field.name()))
+ }
+ Predicate::And(children) | Predicate::Or(children) => children
+ .iter()
+ .any(|child| predicate_references_any_field(child, field_names,
table_fields)),
+ Predicate::Not(inner) => predicate_references_any_field(inner,
field_names, table_fields),
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => false,
+ }
+}
+
+fn collect_blob_view_structs(
+ batch: &RecordBatch,
+ blob_view_fields: &HashSet<String>,
+ view_structs: &mut HashSet<BlobViewStruct>,
+) -> crate::Result<()> {
+ for (idx, field) in batch.schema().fields().iter().enumerate() {
+ if !blob_view_fields.contains(field.name()) {
+ continue;
+ }
+ let col = binary_column(batch, idx, field.name())?;
+ for row in 0..col.len() {
+ if col.is_null(row) {
+ continue;
+ }
+ let value = col.value(row);
+ if !BlobViewStruct::is_blob_view_struct(value) {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "blob-view-field '{}' requires blob field value to be
a serialized BlobViewStruct",
+ field.name()
+ ),
+ source: None,
+ });
+ }
+ view_structs.insert(BlobViewStruct::deserialize(value)?);
+ }
+ }
+ Ok(())
+}
+
+fn replace_blob_view_columns(
+ batch: RecordBatch,
+ blob_view_fields: &HashSet<String>,
+ lookup: &BlobViewLookup,
+) -> crate::Result<RecordBatch> {
+ let schema = batch.schema();
+ let mut columns: Vec<Arc<dyn arrow_array::Array>> =
Vec::with_capacity(batch.num_columns());
+ let mut changed = false;
+
+ for (idx, field) in schema.fields().iter().enumerate() {
+ if !blob_view_fields.contains(field.name()) {
+ columns.push(batch.column(idx).clone());
+ continue;
+ }
+
+ let col = binary_column(&batch, idx, field.name())?;
+ let mut builder = arrow_array::builder::BinaryBuilder::new();
+ for row in 0..col.len() {
+ if col.is_null(row) {
+ builder.append_null();
+ continue;
+ }
+
+ let value = col.value(row);
+ if !BlobViewStruct::is_blob_view_struct(value) {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "blob-view-field '{}' requires blob field value to be
a serialized BlobViewStruct",
+ field.name()
+ ),
+ source: None,
+ });
+ }
+ let view_struct = BlobViewStruct::deserialize(value)?;
+ match lookup.descriptor(&view_struct)? {
+ None => builder.append_null(),
+ Some(descriptor) =>
builder.append_value(descriptor.serialize()),
+ }
+ }
+ columns.push(Arc::new(builder.finish()));
+ changed = true;
+ }
+
+ if !changed {
+ return Ok(batch);
+ }
+
+ RecordBatch::try_new(schema, columns).map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to rebuild RecordBatch after resolving blob
views: {e}"),
+ source: Some(Box::new(e)),
+ })
+}
+
+fn binary_column<'a>(
+ batch: &'a RecordBatch,
+ idx: usize,
+ field_name: &str,
+) -> crate::Result<&'a BinaryArray> {
+ batch
+ .column(idx)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!("blob-view-field '{field_name}' requires a
BinaryArray column"),
+ source: None,
+ })
+}
+
+#[derive(Debug, Default)]
+struct BlobViewLookup {
+ descriptors: HashMap<BlobViewStruct, Option<BlobDescriptor>>,
+}
+
+impl BlobViewLookup {
+ async fn load(rest_env: RESTEnv, view_structs: HashSet<BlobViewStruct>) ->
crate::Result<Self> {
+ if view_structs.is_empty() {
+ return Ok(Self::default());
+ }
+
+ let mut by_table_and_field: HashMap<
+ (crate::catalog::Identifier, i32),
+ Vec<BlobViewStruct>,
+ > = HashMap::new();
+ for view_struct in view_structs {
+ by_table_and_field
+ .entry((view_struct.identifier().clone(),
view_struct.field_id()))
+ .or_default()
+ .push(view_struct);
+ }
+
+ let mut lookup = Self::default();
+ for ((identifier, field_id), refs) in by_table_and_field {
+ let table = rest_env.get_table(&identifier).await?;
+ let field = table
+ .schema()
+ .fields()
+ .iter()
+ .find(|field| field.id() == field_id)
+ .cloned()
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!(
+ "Cannot find blob field id {field_id} in upstream
table {}",
+ identifier.full_name()
+ ),
+ source: None,
+ })?;
+ if !field.data_type().is_blob_type() {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Field id {field_id} in upstream table {} is not a
BLOB field",
+ identifier.full_name()
+ ),
+ source: None,
+ });
+ }
+
+ let mut options = HashMap::new();
+ options.insert("blob-as-descriptor".to_string(),
"true".to_string());
+ let table = table.copy_with_options(options);
+ let row_ranges = row_ranges_for_blob_view_refs(&refs);
+ let mut read_builder = table.new_read_builder();
+ read_builder.with_read_type(vec![field.clone(), row_id_field()]);
+ read_builder.with_row_ranges(row_ranges);
+ let plan = read_builder.new_scan().plan().await?;
+ let read = read_builder.new_read()?;
+ let mut stream = read.to_arrow(plan.splits())?;
+
+ while let Some(batch) = stream.next().await {
+ let batch = batch?;
+ let blob_col = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!(
+ "Upstream blob field '{}' did not read as
BinaryArray",
+ field.name()
+ ),
+ source: None,
+ })?;
+ let row_id_col = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Upstream _ROW_ID did not read as
Int64Array".to_string(),
+ source: None,
+ })?;
+
+ for row in 0..batch.num_rows() {
+ if row_id_col.is_null(row) {
+ continue;
+ }
+ let view_struct =
+ BlobViewStruct::new(identifier.clone(), field_id,
row_id_col.value(row));
+ if blob_col.is_null(row) {
+ lookup.descriptors.insert(view_struct, None);
+ continue;
+ }
+ let value = blob_col.value(row);
+ if !BlobDescriptor::is_blob_descriptor(value) {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "BlobViewStruct {} field_id={} row_id={}
resolved to non-BlobDescriptor bytes",
+ identifier.full_name(),
+ field_id,
+ row_id_col.value(row)
+ ),
+ source: None,
+ });
+ }
+ lookup
+ .descriptors
+ .insert(view_struct,
Some(BlobDescriptor::deserialize(value)?));
+ }
+ }
+
+ for view_struct in refs {
+ if !lookup.descriptors.contains_key(&view_struct) {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "BlobViewStruct not found in upstream table:
identifier={}, field_id={}, row_id={}",
+ view_struct.identifier().full_name(),
+ view_struct.field_id(),
+ view_struct.row_id()
+ ),
+ source: None,
+ });
+ }
+ }
+ }
+
+ Ok(lookup)
+ }
+
+ fn descriptor(&self, view_struct: &BlobViewStruct) ->
crate::Result<Option<&BlobDescriptor>> {
+ self.descriptors
+ .get(view_struct)
+ .map(Option::as_ref)
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!(
+ "BlobViewStruct not found in preloaded cache:
identifier={}, field_id={}, row_id={}",
+ view_struct.identifier().full_name(),
+ view_struct.field_id(),
+ view_struct.row_id()
+ ),
+ source: None,
+ })
+ }
+}
+
+fn row_ranges_for_blob_view_refs(refs: &[BlobViewStruct]) -> Vec<RowRange> {
+ let mut row_ids =
refs.iter().map(BlobViewStruct::row_id).collect::<Vec<_>>();
+ row_ids.sort_unstable();
+ row_ids.dedup();
+
+ let mut ranges = Vec::new();
+ let mut iter = row_ids.into_iter();
+ let Some(mut start) = iter.next() else {
+ return ranges;
+ };
+ let mut end = start;
+ for row_id in iter {
+ if end.checked_add(1) == Some(row_id) {
+ end = row_id;
+ } else {
+ ranges.push(RowRange::new(start, end));
+ start = row_id;
+ end = row_id;
+ }
+ }
+ ranges.push(RowRange::new(start, end));
+ ranges
+}
+
+fn row_id_field() -> DataField {
+ DataField::new(
+ ROW_ID_FIELD_ID,
+ ROW_ID_FIELD_NAME.to_string(),
+ DataType::BigInt(BigIntType::with_nullable(true)),
+ )
+}
+
#[allow(clippy::too_many_arguments)]
fn open_source_stream(
split: &DataSplit,
diff --git a/crates/paimon/src/table/dedicated_format_file_writer.rs
b/crates/paimon/src/table/dedicated_format_file_writer.rs
index 347d435..15f8803 100644
--- a/crates/paimon/src/table/dedicated_format_file_writer.rs
+++ b/crates/paimon/src/table/dedicated_format_file_writer.rs
@@ -16,10 +16,9 @@
// under the License.
use crate::io::FileIO;
-use crate::spec::{BlobDescriptor, DataField, DataFileMeta, DataType};
+use crate::spec::{BlobViewStruct, DataField, DataFileMeta, DataType};
use crate::table::data_file_writer::DataFileWriter;
use crate::Result;
-use arrow_array::builder::BinaryBuilder;
use arrow_array::{Array, RecordBatch};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
@@ -57,6 +56,7 @@ pub(crate) struct AppendDedicatedFormatFileWriter {
vector_writer: Option<VectorFieldWriter>,
normal_column_indices: Vec<usize>,
normal_schema: Arc<arrow_schema::Schema>,
+ blob_view_column_indices: Vec<(usize, String)>,
}
impl AppendDedicatedFormatFileWriter {
@@ -78,7 +78,8 @@ impl AppendDedicatedFormatFileWriter {
input_schema: &arrow_schema::Schema,
table_fields: &[DataField],
format_options: &HashMap<String, String>,
- blob_descriptor_fields: &HashSet<String>,
+ blob_inline_fields: &HashSet<String>,
+ blob_view_fields: &HashSet<String>,
) -> Self {
let mut normal_column_indices = Vec::new();
let mut normal_arrow_fields = Vec::new();
@@ -91,7 +92,7 @@ impl AppendDedicatedFormatFileWriter {
for (idx, field) in table_fields.iter().enumerate() {
let is_blob = field.data_type().is_blob_type();
- let is_descriptor = blob_descriptor_fields.contains(field.name());
+ let is_inline = blob_inline_fields.contains(field.name());
let is_dedicated_vector =
vector_file_format.is_some() && matches!(field.data_type(),
DataType::Vector(_));
@@ -100,7 +101,7 @@ impl AppendDedicatedFormatFileWriter {
vector_arrow_fields.push(input_schema.field(idx).clone());
vector_table_fields.push(field.clone());
vector_field_names.push(field.name().to_string());
- } else if is_blob && !is_descriptor {
+ } else if is_blob && !is_inline {
blob_writers.push(BlobFieldWriter {
writer: DataFileWriter::new(
file_io.clone(),
@@ -186,6 +187,12 @@ impl AppendDedicatedFormatFileWriter {
vector_writer,
normal_column_indices,
normal_schema,
+ blob_view_column_indices: table_fields
+ .iter()
+ .enumerate()
+ .filter(|(_, field)| blob_view_fields.contains(field.name()))
+ .map(|(idx, field)| (idx, field.name().to_string()))
+ .collect(),
}
}
@@ -194,6 +201,8 @@ impl AppendDedicatedFormatFileWriter {
return Ok(());
}
+ self.validate_blob_view_columns(batch)?;
+
// Write normal columns
let normal_columns: Vec<Arc<dyn arrow_array::Array>> = self
.normal_column_indices
@@ -250,6 +259,47 @@ impl AppendDedicatedFormatFileWriter {
Ok(())
}
+ fn validate_blob_view_columns(&self, batch: &RecordBatch) -> Result<()> {
+ for (column_index, field_name) in &self.blob_view_column_indices {
+ let Some(col) = batch
+ .column(*column_index)
+ .as_any()
+ .downcast_ref::<arrow_array::BinaryArray>()
+ else {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "blob-view-field '{field_name}' requires a BinaryArray
value column"
+ ),
+ source: None,
+ });
+ };
+ for row in 0..col.len() {
+ if col.is_null(row) {
+ continue;
+ }
+ let value = col.value(row);
+ if !BlobViewStruct::is_blob_view_struct(value) {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "blob-view-field '{field_name}' requires blob
field value to be a serialized BlobViewStruct"
+ ),
+ source: None,
+ });
+ }
+ let view = BlobViewStruct::deserialize(value)?;
+ if view.serialize()?.as_slice() != value {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "blob-view-field '{field_name}' contains a
non-canonical BlobViewStruct payload"
+ ),
+ source: None,
+ });
+ }
+ }
+ }
+ Ok(())
+ }
+
pub(crate) async fn prepare_commit(&mut self) -> Result<Vec<DataFileMeta>>
{
let mut results = self.normal_writer.prepare_commit().await?;
@@ -266,53 +316,3 @@ impl AppendDedicatedFormatFileWriter {
Ok(results)
}
}
-
-/// For each row in a blob column, if the value is a serialized
`BlobDescriptor`,
-/// resolve it by reading the actual data from the referenced
URI+offset+length.
-/// Raw data values are passed through unchanged.
-pub(crate) async fn resolve_blob_column(
- col: &arrow_array::BinaryArray,
- file_io: &FileIO,
-) -> Result<arrow_array::BinaryArray> {
- use crate::io::FileRead;
- use std::collections::HashMap;
-
- let mut needs_resolve = false;
- for i in 0..col.len() {
- if !col.is_null(i) && BlobDescriptor::is_blob_descriptor(col.value(i))
{
- needs_resolve = true;
- break;
- }
- }
-
- if !needs_resolve {
- return Ok(col.clone());
- }
-
- let mut readers: HashMap<String, Box<dyn FileRead>> = HashMap::new();
- let mut builder = BinaryBuilder::with_capacity(col.len(), 0);
- for i in 0..col.len() {
- if col.is_null(i) {
- builder.append_null();
- } else {
- let value = col.value(i);
- if BlobDescriptor::is_blob_descriptor(value) {
- let desc = BlobDescriptor::deserialize(value)?;
- let uri = desc.uri().to_string();
- if !readers.contains_key(&uri) {
- let input = file_io.new_input(&uri)?;
- let reader = input.reader().await?;
- readers.insert(uri.clone(), Box::new(reader));
- }
- let reader = readers.get(&uri).unwrap();
- let start = desc.offset() as u64;
- let end = start + desc.length() as u64;
- let data = reader.read(start..end).await?;
- builder.append_value(&data);
- } else {
- builder.append_value(value);
- }
- }
- }
- Ok(builder.finish())
-}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 28e3a6a..1865593 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -20,6 +20,7 @@
pub(crate) mod aggregator;
pub(crate) mod bin_pack;
mod bitmap_global_index_reader;
+mod blob_resolver;
mod branch_manager;
mod btree_global_index_build_builder;
mod btree_global_index_drop_builder;
diff --git a/crates/paimon/src/table/rest_env.rs
b/crates/paimon/src/table/rest_env.rs
index e7b1107..cb52f2b 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -15,11 +15,18 @@
// specific language governing permissions and limitations
// under the License.
-//! REST environment for creating RESTSnapshotCommit instances.
+//! REST environment for REST-backed table operations.
use crate::api::rest_api::RESTApi;
-use crate::catalog::Identifier;
+use crate::api::rest_error::RestError;
+use crate::catalog::{Identifier, RESTTokenFileIO};
+use crate::common::Options;
+use crate::error::Error;
+use crate::io::FileIO;
+use crate::spec::TableSchema;
use crate::table::snapshot_commit::{RESTSnapshotCommit, SnapshotCommit};
+use crate::table::Table;
+use crate::Result;
use std::sync::Arc;
/// REST environment that holds the REST API client, identifier, and uuid
@@ -29,6 +36,8 @@ pub struct RESTEnv {
identifier: Identifier,
uuid: String,
api: Arc<RESTApi>,
+ options: Options,
+ data_token_enabled: bool,
}
impl std::fmt::Debug for RESTEnv {
@@ -42,11 +51,19 @@ impl std::fmt::Debug for RESTEnv {
impl RESTEnv {
/// Create a new RESTEnv.
- pub fn new(identifier: Identifier, uuid: String, api: Arc<RESTApi>) ->
Self {
+ pub fn new(
+ identifier: Identifier,
+ uuid: String,
+ api: Arc<RESTApi>,
+ options: Options,
+ data_token_enabled: bool,
+ ) -> Self {
Self {
identifier,
uuid,
api,
+ options,
+ data_token_enabled,
}
}
@@ -60,6 +77,85 @@ impl RESTEnv {
&self.identifier
}
+ /// Load a table through the same REST catalog environment.
+ pub async fn get_table(&self, identifier: &Identifier) -> Result<Table> {
+ Self::load_table(
+ identifier,
+ self.api.clone(),
+ self.options.clone(),
+ self.data_token_enabled,
+ )
+ .await
+ }
+
+ /// Load a REST table and attach a fresh RESTEnv to it.
+ pub(crate) async fn load_table(
+ identifier: &Identifier,
+ api: Arc<RESTApi>,
+ options: Options,
+ data_token_enabled: bool,
+ ) -> Result<Table> {
+ let response = api
+ .get_table(identifier)
+ .await
+ .map_err(|e| map_rest_error_for_table(e, identifier))?;
+
+ let schema = response.schema.ok_or_else(|| Error::DataInvalid {
+ message: format!("Table {} response missing schema",
identifier.full_name()),
+ source: None,
+ })?;
+
+ let schema_id = response.schema_id.ok_or_else(|| Error::DataInvalid {
+ message: format!(
+ "Table {} response missing schema_id",
+ identifier.full_name()
+ ),
+ source: None,
+ })?;
+ let table_schema = TableSchema::new(schema_id, &schema);
+
+ let table_path = response.path.ok_or_else(|| Error::DataInvalid {
+ message: format!("Table {} response missing path",
identifier.full_name()),
+ source: None,
+ })?;
+
+ let is_external = response.is_external.ok_or_else(||
Error::DataInvalid {
+ message: format!(
+ "Table {} response missing is_external",
+ identifier.full_name()
+ ),
+ source: None,
+ })?;
+
+ let uuid = response.id.ok_or_else(|| Error::DataInvalid {
+ message: format!(
+ "Table {} response missing id (uuid)",
+ identifier.full_name()
+ ),
+ source: None,
+ })?;
+
+ let file_io = if data_token_enabled && !is_external {
+ RESTTokenFileIO::new(identifier.clone(), table_path.clone(),
options.clone())
+ .build_file_io()
+ .await?
+ } else {
+ let mut builder = FileIO::from_path(&table_path)?;
+ builder = builder.with_props(options.to_map());
+ builder.build()?
+ };
+
+ let rest_env = RESTEnv::new(identifier.clone(), uuid, api, options,
data_token_enabled);
+
+ Ok(Table::new(
+ file_io,
+ identifier.clone(),
+ table_path,
+ table_schema,
+ Some(rest_env),
+ ))
+ }
+
/// Create a `RESTSnapshotCommit` from this environment.
pub fn snapshot_commit(&self) -> Arc<dyn SnapshotCommit> {
Arc::new(RESTSnapshotCommit::new(
@@ -69,3 +165,19 @@ impl RESTEnv {
))
}
}
+
+fn map_rest_error_for_table(err: Error, identifier: &Identifier) -> Error {
+ match err {
+ Error::RestApi {
+ source: RestError::NoSuchResource { .. },
+ } => Error::TableNotExist {
+ full_name: identifier.full_name(),
+ },
+ Error::RestApi {
+ source: RestError::AlreadyExists { .. },
+ } => Error::TableAlreadyExist {
+ full_name: identifier.full_name(),
+ },
+ other => other,
+ }
+}
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 216fbd6..29e8adf 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -200,6 +200,9 @@ impl<'a> TableRead<'a> {
self.data_predicates.clone(),
core_options.blob_as_descriptor(),
core_options.blob_descriptor_fields(),
+ core_options.blob_view_fields(),
+ core_options.blob_view_resolve_enabled(),
+ self.table.rest_env().cloned(),
)?;
reader.read(data_splits)
}
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 1854c53..034e634 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -111,8 +111,10 @@ pub struct TableWrite {
bucket_assigner: BucketAssignerEnum,
/// Whether this is an overwrite operation (skip seq/index restore).
is_overwrite: bool,
- /// Blob descriptor fields (stored inline in parquet, not as separate
.blob files).
- blob_descriptor_fields: HashSet<String>,
+ /// Blob view fields (stored inline as BlobViewStruct bytes).
+ blob_view_fields: HashSet<String>,
+ /// Descriptor + view inline blob fields.
+ blob_inline_fields: HashSet<String>,
/// Whether the table has non-descriptor blob fields requiring a
dedicated-format writer.
has_blob_fields: bool,
/// Dedicated vector-store file format, when configured.
@@ -127,19 +129,21 @@ impl TableWrite {
let schema = table.schema();
let core_options = CoreOptions::new(schema.options());
let blob_descriptor_fields = core_options.blob_descriptor_fields();
+ let blob_view_fields = core_options.blob_view_fields();
+ let blob_inline_fields = core_options.blob_inline_fields();
- for name in &blob_descriptor_fields {
+ for name in &blob_inline_fields {
match schema.fields().iter().find(|f| f.name() == name) {
None => {
return Err(crate::Error::DataInvalid {
- message: format!("blob-descriptor-field '{name}' does
not exist in schema"),
+ message: format!("inline blob field '{name}' does not
exist in schema"),
source: None,
});
}
Some(f) if !f.data_type().is_blob_type() => {
return Err(crate::Error::DataInvalid {
message: format!(
- "blob-descriptor-field '{name}' is not a top-level
BLOB field"
+ "inline blob field '{name}' is not a top-level
BLOB field"
),
source: None,
});
@@ -147,6 +151,18 @@ impl TableWrite {
_ => {}
}
}
+ if let Some(name) = blob_descriptor_fields
+ .intersection(&blob_view_fields)
+ .next()
+ .cloned()
+ {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Field '{name}' in 'blob-view-field' can not also be in
'blob-descriptor-field'"
+ ),
+ source: None,
+ });
+ }
let total_buckets = core_options.bucket();
let has_primary_keys = !schema.primary_keys().is_empty();
@@ -313,7 +329,7 @@ impl TableWrite {
let has_blob_fields = schema
.fields()
.iter()
- .any(|f| f.data_type().is_blob_type() &&
!blob_descriptor_fields.contains(f.name()));
+ .any(|f| f.data_type().is_blob_type() &&
!blob_inline_fields.contains(f.name()));
let has_dedicated_vector_fields = vector_file_format.is_some()
&& schema
.fields()
@@ -345,7 +361,8 @@ impl TableWrite {
commit_user,
bucket_assigner,
is_overwrite,
- blob_descriptor_fields,
+ blob_view_fields,
+ blob_inline_fields,
has_blob_fields,
vector_file_format,
has_dedicated_vector_fields,
@@ -641,7 +658,10 @@ impl TableWrite {
/// Create an append-only writer for non-PK tables.
fn create_append_writer(&self, partition_path: String, bucket: i32) ->
Result<FileWriter> {
- if self.has_blob_fields || self.has_dedicated_vector_fields {
+ if self.has_blob_fields
+ || self.has_dedicated_vector_fields
+ || !self.blob_view_fields.is_empty()
+ {
let fields = self.table.schema().fields();
let input_schema = build_target_arrow_schema(fields)?;
Ok(FileWriter::AppendDedicated(Box::new(
@@ -662,7 +682,8 @@ impl TableWrite {
&input_schema,
fields,
self.table.schema().options(),
- &self.blob_descriptor_fields,
+ &self.blob_inline_fields,
+ &self.blob_view_fields,
),
)))
} else {
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 70a86c5..0a2e20f 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -21,11 +21,19 @@
//! through the Catalog trait interface.
use std::collections::HashMap;
+use std::sync::Arc;
+use arrow_array::{Array, BinaryArray, Int32Array, RecordBatch, StringArray};
+use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as
ArrowSchema};
+use futures::TryStreamExt;
use paimon::api::ConfigResponse;
use paimon::catalog::{Catalog, Identifier, RESTCatalog};
use paimon::common::Options;
-use paimon::spec::{BigIntType, DataType, Schema, SchemaChange, VarCharType};
+use paimon::spec::{
+ BigIntType, BlobType, BlobViewStruct, DataType, Datum, IntType,
PredicateBuilder, Schema,
+ SchemaChange, VarCharType,
+};
+use paimon::{CatalogOptions, FileSystemCatalog, Table};
mod mock_server;
use mock_server::{start_mock_server, RESTServer};
@@ -75,6 +83,89 @@ fn test_schema() -> Schema {
.expect("Failed to build schema")
}
+fn blob_schema(options: &[(&str, &str)]) -> Schema {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::new(255).unwrap()))
+ .column("picture", DataType::Blob(BlobType::new()))
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().expect("Failed to build blob schema")
+}
+
+fn blob_batch(ids: Vec<i32>, names: Vec<&str>, pictures: Vec<Vec<u8>>) ->
RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("name", ArrowDataType::Utf8, true),
+ ArrowField::new("picture", ArrowDataType::Binary, true),
+ ]));
+ let picture_refs = pictures
+ .iter()
+ .map(|bytes| Some(bytes.as_slice()))
+ .collect::<Vec<_>>();
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(StringArray::from(names)),
+ Arc::new(BinaryArray::from(picture_refs)),
+ ],
+ )
+ .unwrap()
+}
+
+async fn write_batch(table: &Table, batch: RecordBatch, commit_user: &str) {
+ let write_builder = table
+ .new_write_builder()
+ .with_commit_user(commit_user)
+ .expect("valid commit user");
+ let mut write = write_builder.new_write().expect("create writer");
+ write.write_arrow_batch(&batch).await.expect("write batch");
+ let messages = write.prepare_commit().await.expect("prepare commit");
+ write_builder
+ .new_commit()
+ .commit(messages)
+ .await
+ .expect("commit batch");
+}
+
+fn collect_blob_rows(batches: &[RecordBatch]) -> Vec<(i32, String,
Option<Vec<u8>>)> {
+ let mut rows = Vec::new();
+ for batch in batches {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let names = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let pictures = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ for row in 0..batch.num_rows() {
+ rows.push((
+ ids.value(row),
+ names.value(row).to_string(),
+ if pictures.is_null(row) {
+ None
+ } else {
+ Some(pictures.value(row).to_vec())
+ },
+ ));
+ }
+ }
+ rows.sort_by_key(|(id, _, _)| *id);
+ rows
+}
+
// ==================== Database Tests ====================
#[tokio::test]
@@ -254,6 +345,150 @@ async fn test_catalog_get_table() {
assert!(table.is_ok(), "failed to get table: {table:?}");
}
+#[tokio::test]
+async fn test_rest_env_get_table_reuses_catalog_environment() {
+ let ctx = setup_catalog(vec!["default"]).await;
+
+ ctx.server.add_table_with_schema(
+ "default",
+ "current_table",
+ test_schema(),
+ "file:///tmp/test_warehouse/default.db/current_table",
+ );
+ ctx.server.add_table_with_schema(
+ "default",
+ "upstream_table",
+ test_schema(),
+ "file:///tmp/test_warehouse/default.db/upstream_table",
+ );
+
+ let current = ctx
+ .catalog
+ .get_table(&Identifier::new("default", "current_table"))
+ .await
+ .expect("current table should load");
+ let upstream = current
+ .rest_env()
+ .expect("REST table should carry RESTEnv")
+ .get_table(&Identifier::new("default", "upstream_table"))
+ .await
+ .expect("upstream table should load via RESTEnv");
+
+ assert_eq!(upstream.identifier().full_name(), "default.upstream_table");
+ assert!(upstream.rest_env().is_some());
+}
+
+// This regression uses FileSystemCatalog to write real table files before
reading
+// them back through RESTCatalog. FileSystemCatalog directory listing is
skipped
+// on Windows elsewhere for the same opendal `fs` StripPrefixError.
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_blob_view_prescan_filters_invalid_filtered_out_reference() {
+ let tmp = tempfile::tempdir().unwrap();
+ let warehouse = format!("file://{}", tmp.path().display());
+
+ let mut fs_options = Options::new();
+ fs_options.set(CatalogOptions::WAREHOUSE, &warehouse);
+ let fs_catalog = FileSystemCatalog::new(fs_options).expect("create
filesystem catalog");
+ fs_catalog
+ .create_database("default", true, HashMap::new())
+ .await
+ .unwrap();
+
+ let source_id = Identifier::new("default", "blob_source");
+ let source_schema = blob_schema(&[]);
+ fs_catalog
+ .create_table(&source_id, source_schema.clone(), false)
+ .await
+ .unwrap();
+ let source = fs_catalog.get_table(&source_id).await.unwrap();
+ write_batch(
+ &source,
+ blob_batch(
+ vec![1, 2],
+ vec!["Alice", "Bob"],
+ vec![b"alice".to_vec(), b"bob".to_vec()],
+ ),
+ "source-writer",
+ )
+ .await;
+
+ let view_id = Identifier::new("default", "blob_view_target");
+ let view_schema = blob_schema(&[("blob-view-field", "picture")]);
+ fs_catalog
+ .create_table(&view_id, view_schema.clone(), false)
+ .await
+ .unwrap();
+ let view = fs_catalog.get_table(&view_id).await.unwrap();
+
+ let picture_field_id = source
+ .schema()
+ .fields()
+ .iter()
+ .find(|field| field.name() == "picture")
+ .unwrap()
+ .id();
+ let filtered_out_bad_ref = BlobViewStruct::new(source_id.clone(),
picture_field_id, 99)
+ .serialize()
+ .unwrap();
+ let kept_ref = BlobViewStruct::new(source_id.clone(), picture_field_id, 1)
+ .serialize()
+ .unwrap();
+ write_batch(
+ &view,
+ blob_batch(
+ vec![1, 2],
+ vec!["Filtered", "Kept"],
+ vec![filtered_out_bad_ref, kept_ref],
+ ),
+ "view-writer",
+ )
+ .await;
+
+ let prefix = "mock-test";
+ let mut defaults = HashMap::new();
+ defaults.insert("prefix".to_string(), prefix.to_string());
+ let server = start_mock_server(
+ "test_warehouse".to_string(),
+ warehouse.clone(),
+ ConfigResponse::new(defaults),
+ vec!["default".to_string()],
+ )
+ .await;
+ server.add_table_with_schema("default", "blob_source", source_schema,
source.location());
+ server.add_table_with_schema("default", "blob_view_target", view_schema,
view.location());
+
+ let url = server.url().expect("Failed to get server URL");
+ let mut rest_options = Options::new();
+ rest_options.set("uri", &url);
+ rest_options.set("warehouse", "test_warehouse");
+ rest_options.set("token.provider", "bear");
+ rest_options.set("token", "test_token");
+ let rest_catalog = RESTCatalog::new(rest_options, true)
+ .await
+ .expect("create rest catalog");
+
+ let rest_view = rest_catalog.get_table(&view_id).await.unwrap();
+ let predicate = PredicateBuilder::new(rest_view.schema().fields())
+ .equal("id", Datum::Int(2))
+ .unwrap();
+ let mut read_builder = rest_view.new_read_builder();
+ read_builder.with_filter(predicate);
+ let plan = read_builder.new_scan().plan().await.unwrap();
+ let read = read_builder.new_read().unwrap();
+ let batches = read
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(
+ collect_blob_rows(&batches),
+ vec![(2, "Kept".to_string(), Some(b"bob".to_vec()))]
+ );
+}
+
#[tokio::test]
async fn test_catalog_get_table_not_found() {
let ctx = setup_catalog(vec!["default"]).await;
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 934eb27..86cf612 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -78,7 +78,13 @@ async fn example() -> Result<(), Box<dyn std::error::Error>>
{
}
```
-`SQLContext::new` creates a session context with the Paimon relation planner
pre-registered. Use `register_catalog(...).await` to add one or more Paimon
catalogs; registering a catalog also registers the built-in table-valued
functions (`vector_search`, `hybrid_search`, and `full_text_search` when the
`fulltext` feature is enabled) against it. It also manages session-scoped
dynamic options internally for `SET`/`RESET` support.
+`SQLContext::new` creates a session context with the Paimon relation planner
+pre-registered. Use `register_catalog(...).await` to add one or more Paimon
+catalogs; registering a catalog also registers the built-in scalar function
+`blob_view` (alias `sys.blob_view`) and the built-in table-valued functions
+(`vector_search`, `hybrid_search`, and `full_text_search` when the `fulltext`
+feature is enabled) against it. It also manages session-scoped dynamic options
+internally for `SET`/`RESET` support.
## Data Types
@@ -111,6 +117,65 @@ columns; DataFusion reads those as Arrow
`FixedSizeList<Float32>`, and vindex
index creation uses `N` as the vector dimension. `SHOW CREATE TABLE` currently
does not round-trip `VECTOR` columns.
+### Blob View
+
+Blob View stores an inline reference to a BLOB value in another table, using a
+Java-compatible `BlobViewStruct` payload. It is useful when one table should
+point at media or large binary content owned by an upstream table without
+copying the bytes at write time.
+
+Declare Blob View columns as top-level BLOB columns and list them in the
+`blob-view-field` table option:
+
+```sql
+CREATE TABLE paimon.my_db.asset_refs (
+ id INT,
+ picture BLOB
+) WITH (
+ 'data-evolution.enabled' = 'true',
+ 'row-tracking.enabled' = 'true',
+ 'blob-view-field' = 'picture'
+);
+```
+
+Use `blob_view(table, field_name_or_id, row_id)` or `sys.blob_view(...)` to
+create the reference. The table argument may be `table`, `database.table`, or
+`catalog.database.table`; the stored reference contains the resolved
+`database.table`, field id, and row id. In typical SQL, read `_ROW_ID` from a
+row-tracking source table:
+
+```sql
+CREATE TABLE paimon.my_db.assets (
+ id INT,
+ picture BLOB
+) WITH (
+ 'data-evolution.enabled' = 'true',
+ 'row-tracking.enabled' = 'true'
+);
+
+INSERT INTO paimon.my_db.asset_refs (id, picture)
+SELECT
+ id,
+ sys.blob_view('my_db.assets', 'picture', "_ROW_ID")
+FROM paimon.my_db.assets;
+```
+
+By default, RESTCatalog-backed reads resolve Blob View fields to the upstream
+BLOB value by reusing the table's REST environment. Other catalog types
+currently preserve the raw serialized `BlobViewStruct` bytes. Set the dynamic
+option `paimon.blob-view.resolve.enabled` to `false` to preserve raw references
+even for RESTCatalog-backed reads:
+
+```sql
+SET 'paimon.blob-view.resolve.enabled' = 'false';
+SELECT id, picture FROM paimon.my_db.asset_refs;
+RESET 'paimon.blob-view.resolve.enabled';
+```
+
+Like ordinary BLOB reads, `paimon.blob-as-descriptor = true` makes resolved
Blob
+View columns return serialized BLOB descriptors instead of loading the BLOB
+bytes.
+
### Variant Usage
`VARIANT` stores semi-structured data using the same logical value + metadata
binary shape as Paimon Java. Use it for JSON-like fields whose schema may
differ row by row.
@@ -1171,6 +1236,15 @@ SELECT * FROM paimon.my_db.assets;
RESET 'paimon.blob-as-descriptor';
```
+Example — preserve Blob View references instead of resolving upstream BLOB
+values on RESTCatalog-backed reads:
+
+```sql
+SET 'paimon.blob-view.resolve.enabled' = 'false';
+SELECT * FROM paimon.my_db.asset_refs;
+RESET 'paimon.blob-view.resolve.enabled';
+```
+
## Temporary Tables
You can register in-memory temporary tables under any catalog. Temporary
tables exist only for the lifetime of the `SQLContext` instance and are
automatically cleaned up when the context is dropped.
