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 f900b02c fix(core): preserve REST Blob view dependency context (#1085)
f900b02c is described below
commit f900b02cd60a3fd04f28fcf8805d73f6892df947
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 9 17:54:26 2026 +0800
fix(core): preserve REST Blob view dependency context (#1085)
---
crates/paimon/src/api/rest_api.rs | 1 +
crates/paimon/src/table/data_evolution_reader.rs | 332 +++++++++++++++++------
crates/paimon/src/table/rest_env.rs | 86 ++++++
crates/paimon/src/table/table_read.rs | 27 +-
crates/paimon/tests/mock_server.rs | 31 ++-
crates/paimon/tests/rest_catalog_test.rs | 23 +-
6 files changed, 390 insertions(+), 110 deletions(-)
diff --git a/crates/paimon/src/api/rest_api.rs
b/crates/paimon/src/api/rest_api.rs
index b2d9a29f..57828e34 100644
--- a/crates/paimon/src/api/rest_api.rs
+++ b/crates/paimon/src/api/rest_api.rs
@@ -104,6 +104,7 @@ pub struct RESTApi {
impl RESTApi {
// Constants for query parameters and headers
+ pub const READ_VIA_HEADER: &'static str = "X-Paimon-Read-Via";
pub const HEADER_PREFIX: &'static str = "header.";
pub const MAX_RESULTS: &'static str = "maxResults";
pub const PAGE_TOKEN: &'static str = "pageToken";
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 933b41b9..b856bbc8 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -820,15 +820,21 @@ impl DataEvolutionReader {
.load_missing(rest_env, &batch, &self.blob_view_fields)
.await?;
}
- batch = self.resolve_blob_view_columns(batch,
blob_view_lookup.as_ref())?;
+ batch = self
+ .resolve_blob_view_columns(batch, blob_view_lookup.as_ref())
+ .await?;
let mut batch = if !self.blob_as_descriptor &&
!descriptor_fields.is_empty() {
- resolve_descriptor_columns(
- batch,
- descriptor_fields,
- &self.file_io,
- &self.blob_read_limiter,
- )
- .await?
+ // View payloads were resolved with their upstream table's FileIO.
+ let fields = if blob_view_lookup.is_some() {
+ descriptor_fields
+ .difference(&self.blob_view_fields)
+ .cloned()
+ .collect()
+ } else {
+ descriptor_fields.clone()
+ };
+ resolve_descriptor_columns(batch, &fields, &self.file_io,
&self.blob_read_limiter)
+ .await?
} else {
batch
};
@@ -928,7 +934,7 @@ impl DataEvolutionReader {
fields
}
- fn resolve_blob_view_columns(
+ async fn resolve_blob_view_columns(
&self,
batch: RecordBatch,
lookup: Option<&BlobViewLookup>,
@@ -937,7 +943,14 @@ impl DataEvolutionReader {
return Ok(batch);
};
- replace_blob_view_columns(batch, &self.blob_view_fields, lookup)
+ replace_blob_view_columns(
+ batch,
+ &self.blob_view_fields,
+ lookup,
+ self.blob_as_descriptor,
+ &self.blob_read_limiter,
+ )
+ .await
}
/// Merge multiple logical sources column-wise for data evolution.
@@ -1381,10 +1394,12 @@ fn collect_blob_view_structs(
Ok(())
}
-fn replace_blob_view_columns(
+async fn replace_blob_view_columns(
batch: RecordBatch,
blob_view_fields: &HashSet<String>,
lookup: &BlobViewLookup,
+ blob_as_descriptor: bool,
+ limiter: &BlobReadLimiter,
) -> crate::Result<RecordBatch> {
let schema = batch.schema();
let mut columns: Vec<Arc<dyn arrow_array::Array>> =
Vec::with_capacity(batch.num_columns());
@@ -1397,10 +1412,11 @@ fn replace_blob_view_columns(
}
let col = large_binary_column(&batch, idx, field.name())?;
- let mut builder = arrow_array::builder::LargeBinaryBuilder::new();
- for row in 0..col.len() {
+ let mut values = vec![None; col.len()];
+ let mut by_table: HashMap<crate::catalog::Identifier, Vec<(usize,
Vec<u8>)>> =
+ HashMap::new();
+ for (row, slot) in values.iter_mut().enumerate() {
if col.is_null(row) {
- builder.append_null();
continue;
}
@@ -1416,11 +1432,35 @@ fn replace_blob_view_columns(
}
let view_struct = BlobViewStruct::deserialize(value)?;
match lookup.descriptor(&view_struct)? {
- None => builder.append_null(),
- Some(descriptor) =>
builder.append_value(descriptor.serialize()),
+ None => {}
+ Some(descriptor) if blob_as_descriptor => *slot =
Some(descriptor.serialize()),
+ Some(descriptor) => by_table
+ .entry(view_struct.identifier().clone())
+ .or_default()
+ .push((row, descriptor.serialize())),
+ }
+ }
+ for (identifier, entries) in by_table {
+ let file_io = lookup
+ .file_ios
+ .get(&identifier)
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!("Missing upstream FileIO for {}",
identifier.full_name()),
+ source: None,
+ })?;
+ let descriptors =
+ LargeBinaryArray::from_iter_values(entries.iter().map(|(_,
value)| value));
+ let payloads = super::blob_resolver::resolve_descriptor_column(
+ &descriptors,
+ file_io,
+ limiter.clone(),
+ )
+ .await?;
+ for ((row, _), value) in entries.into_iter().zip(payloads.iter()) {
+ values[row] = value.map(<[u8]>::to_vec);
}
}
- columns.push(Arc::new(builder.finish()));
+ columns.push(Arc::new(LargeBinaryArray::from_iter(values)));
changed = true;
}
@@ -1452,6 +1492,8 @@ fn large_binary_column<'a>(
#[derive(Debug, Default)]
struct BlobViewLookup {
descriptors: HashMap<BlobViewStruct, Option<BlobDescriptor>>,
+ file_ios: HashMap<crate::catalog::Identifier, FileIO>,
+ rest_env: Option<RESTEnv>,
}
/// Resolve view references only after the caller has selected authorized rows.
@@ -1459,14 +1501,17 @@ pub(crate) fn resolve_blob_view_stream(
stream: ArrowRecordBatchStream,
fields: HashSet<String>,
rest_env: RESTEnv,
+ blob_as_descriptor: bool,
+ blob_parallelism: usize,
) -> ArrowRecordBatchStream {
Box::pin(async_stream::try_stream! {
+ let limiter = BlobReadLimiter::with_parallelism(blob_parallelism);
let mut stream = stream;
let mut lookup = BlobViewLookup::default();
while let Some(batch) = stream.next().await {
let batch = batch?;
lookup.load_missing(rest_env.clone(), &batch, &fields).await?;
- yield replace_blob_view_columns(batch, &fields, &lookup)?;
+ yield replace_blob_view_columns(batch, &fields, &lookup,
blob_as_descriptor, &limiter).await?;
}
})
}
@@ -1482,8 +1527,11 @@ impl BlobViewLookup {
collect_blob_view_structs(batch, blob_view_fields, &mut view_structs)?;
view_structs.retain(|view| !self.descriptors.contains_key(view));
if !view_structs.is_empty() {
- let loaded = Self::load(rest_env, view_structs).await?;
+ let loaded =
+ Self::load(self.rest_env.clone().unwrap_or(rest_env),
view_structs).await?;
self.descriptors.extend(loaded.descriptors);
+ self.file_ios.extend(loaded.file_ios);
+ self.rest_env = loaded.rest_env;
}
Ok(())
}
@@ -1493,101 +1541,106 @@ impl BlobViewLookup {
return Ok(Self::default());
}
- let mut by_table_and_field: HashMap<
- (crate::catalog::Identifier, i32),
- Vec<BlobViewStruct>,
- > = HashMap::new();
+ let rest_env = rest_env.for_dependency_reads().await?;
+ let mut by_table: HashMap<crate::catalog::Identifier,
Vec<BlobViewStruct>> = HashMap::new();
for view_struct in view_structs {
- by_table_and_field
- .entry((view_struct.identifier().clone(),
view_struct.field_id()))
+ by_table
+ .entry(view_struct.identifier().clone())
.or_default()
.push(view_struct);
}
- let mut lookup = Self::default();
- for ((identifier, field_id), refs) in by_table_and_field {
+ let mut lookup = Self {
+ rest_env: Some(rest_env.clone()),
+ ..Self::default()
+ };
+ for (identifier, refs) in by_table {
let table = rest_env.get_table(&identifier).await?;
- let field = table
+ lookup
+ .file_ios
+ .insert(identifier.clone(), table.file_io().clone());
+ let field_ids: HashSet<_> =
refs.iter().map(BlobViewStruct::field_id).collect();
+ // Java groups fields and row ranges by table. One projected scan
+ // keeps all referenced fields on the same upstream snapshot.
+ let fields: Vec<_> = table
.schema()
.fields()
.iter()
- .find(|field| field.id() == field_id)
+ .filter(|field| field_ids.contains(&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,
- });
+ .collect();
+ for id in &field_ids {
+ if !fields.iter().any(|field| field.id() == *id) {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Cannot find blob field id {id} in upstream table
{}",
+ identifier.full_name()
+ ),
+ source: None,
+ });
+ }
+ }
+ for field in &fields {
+ if !field.data_type().is_blob_type() {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Field id {} in upstream table {} is not a BLOB
field",
+ field.id(),
+ 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 table = table.copy_with_options(HashMap::from([(
+ "blob-as-descriptor".to_string(),
+ "true".to_string(),
+ )]));
+ let mut read_fields = fields.clone();
+ read_fields.push(crate::spec::row_id_data_field());
let mut read_builder = table.new_read_builder();
- read_builder.with_read_type(vec![field.clone(),
crate::spec::row_id_data_field()]);
- read_builder.with_row_ranges(row_ranges);
+ read_builder.with_read_type(read_fields);
+ read_builder.with_row_ranges(row_ranges_for_blob_view_refs(&refs));
let plan = read_builder.new_scan().plan().await?;
- let read = read_builder.new_read()?;
- let mut stream = read.to_arrow(plan.splits())?;
+ let mut stream = read_builder.new_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::<LargeBinaryArray>()
- .ok_or_else(|| Error::DataInvalid {
- message: format!(
- "Upstream blob field '{}' did not read as
LargeBinaryArray",
- field.name()
- ),
- source: None,
- })?;
let row_id_col = batch
- .column(1)
+ .column(fields.len())
.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,
- });
+ for (index, field) in fields.iter().enumerate() {
+ let blob_col = large_binary_column(&batch, index,
field.name())?;
+ 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),
+ );
+ let descriptor = if blob_col.is_null(row) {
+ None
+ } else {
+ 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,
+ });
+ }
+ Some(BlobDescriptor::deserialize(value)?)
+ };
+ lookup.descriptors.insert(view_struct, descriptor);
}
- lookup
- .descriptors
- .insert(view_struct,
Some(BlobDescriptor::deserialize(value)?));
}
}
@@ -3191,6 +3244,107 @@ mod tests {
use std::path::{Path, PathBuf};
use tempfile::tempdir;
+ #[tokio::test]
+ async fn
blob_view_payloads_use_each_upstream_file_io_even_for_the_same_uri() {
+ let uri = "memory:/referenced-payload";
+ let first_io = FileIO::from_path(uri).unwrap().build().unwrap();
+ let second_io = FileIO::from_path(uri).unwrap().build().unwrap();
+ first_io
+ .new_output(uri)
+ .unwrap()
+ .write(Bytes::from_static(b"first"))
+ .await
+ .unwrap();
+ second_io
+ .new_output(uri)
+ .unwrap()
+ .write(Bytes::from_static(b"other"))
+ .await
+ .unwrap();
+ let first = BlobViewStruct::new(Identifier::new("db", "first"), 1, 0);
+ let second = BlobViewStruct::new(Identifier::new("db", "second"), 1,
0);
+ let null = BlobViewStruct::new(first.identifier().clone(), 1, 1);
+ let descriptor = BlobDescriptor::new(uri.to_string(), 0, 5);
+ let mut lookup = BlobViewLookup::default();
+ lookup
+ .descriptors
+ .insert(first.clone(), Some(descriptor.clone()));
+ lookup
+ .descriptors
+ .insert(second.clone(), Some(descriptor.clone()));
+ lookup.descriptors.insert(null.clone(), None);
+ lookup.file_ios.insert(first.identifier().clone(), first_io);
+ lookup
+ .file_ios
+ .insert(second.identifier().clone(), second_io);
+ let refs = [
+ Some(first.serialize().unwrap()),
+ None,
+ Some(second.serialize().unwrap()),
+ Some(first.serialize().unwrap()),
+ Some(null.serialize().unwrap()),
+ ];
+ let batch = RecordBatch::try_from_iter([(
+ "payload",
+ Arc::new(LargeBinaryArray::from_iter(refs)) as
arrow_array::ArrayRef,
+ )])
+ .unwrap();
+ let fields = HashSet::from(["payload".to_string()]);
+ let result = replace_blob_view_columns(
+ batch.clone(),
+ &fields,
+ &lookup,
+ false,
+ &BlobReadLimiter::new(),
+ )
+ .await
+ .unwrap();
+ let values = large_binary_column(&result, 0, "payload")
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>();
+ assert_eq!(
+ values,
+ vec![
+ Some(b"first".as_slice()),
+ None,
+ Some(b"other".as_slice()),
+ Some(b"first".as_slice()),
+ None
+ ]
+ );
+ // Descriptor output must not open payload files.
+ lookup.file_ios.clear();
+ let result = replace_blob_view_columns(
+ batch.clone(),
+ &fields,
+ &lookup,
+ true,
+ &BlobReadLimiter::new(),
+ )
+ .await
+ .unwrap();
+ let expected = descriptor.serialize();
+ assert_eq!(
+ large_binary_column(&result, 0, "payload")
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>(),
+ vec![
+ Some(expected.as_slice()),
+ None,
+ Some(expected.as_slice()),
+ Some(expected.as_slice()),
+ None
+ ]
+ );
+ let error =
+ replace_blob_view_columns(batch, &fields, &lookup, false,
&BlobReadLimiter::new())
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("Missing upstream FileIO"));
+ }
+
mod blob_test_utils {
include!(concat!(env!("CARGO_MANIFEST_DIR"), "/blob_test_utils.rs"));
}
diff --git a/crates/paimon/src/table/rest_env.rs
b/crates/paimon/src/table/rest_env.rs
index fe811404..48ad5e19 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -204,6 +204,31 @@ impl RESTEnv {
&self.uuid
}
+ /// Attach Java's outermost table identity to dependency requests. The
+ /// dependency client is separate so ordinary requests retain their
headers.
+ pub(crate) async fn for_dependency_reads(&self) -> Result<Self> {
+ let key = format!("{}{}", RESTApi::HEADER_PREFIX,
RESTApi::READ_VIA_HEADER);
+ let mut options = self.api.options().clone();
+ if options.get(&key).is_some() {
+ return Ok(self.clone());
+ }
+ let identifier =
+ serde_json::to_string(&self.identifier).map_err(|error|
Error::DataInvalid {
+ message: "Cannot serialize dependency read
identifier".to_string(),
+ source: Some(Box::new(error)),
+ })?;
+ options.set(
+ key,
+ crate::api::rest_util::RESTUtil::encode_string(&identifier),
+ );
+ let api = Arc::new(RESTApi::new(options.clone(), false).await?);
+ Ok(Self {
+ api,
+ options,
+ ..self.clone()
+ })
+ }
+
/// Load a table through the same REST catalog environment.
pub async fn get_table(&self, identifier: &Identifier) -> Result<Table> {
Self::load_table(
@@ -603,6 +628,67 @@ mod tests {
use crate::common::CatalogOptions;
use crate::io::cache::create_local_cache;
+ #[tokio::test]
+ async fn
dependency_reads_preserve_the_outermost_identifier_and_catalog_options() {
+ let table = crate::table::rest_query_auth_table().await;
+ let mut env = table.rest_env().unwrap().clone();
+ env.identifier = Identifier::new("db 空格", "root$branch_dev");
+ let mut options = env.api.options().clone();
+ options.set("prefix", "configured-prefix");
+ options.set("header.X-Custom", "kept");
+ env.api = Arc::new(RESTApi::new(options, false).await.unwrap());
+ // The API has the merged server configuration, while env.options can
+ // still be the original catalog options.
+ let dependency = env.for_dependency_reads().await.unwrap();
+ let key = "header.X-Paimon-Read-Via";
+ assert!(env.api.options().get(key).is_none());
+ assert!(!Arc::ptr_eq(&env.api, &dependency.api));
+ assert_eq!(
+ dependency.api.options().get("prefix").unwrap(),
+ "configured-prefix"
+ );
+ assert_eq!(
+ dependency.api.options().get("header.X-Custom").unwrap(),
+ "kept"
+ );
+ let decoded =
+
crate::api::RESTUtil::decode_string(dependency.api.options().get(key).unwrap());
+ assert_eq!(
+ decoded,
+ r#"{"database":"db 空格","object":"root$branch_dev"}"#
+ );
+ let mut nested = dependency.clone();
+ nested.identifier = Identifier::new("other", "middle");
+ let nested = nested.for_dependency_reads().await.unwrap();
+ assert!(Arc::ptr_eq(&dependency.api, &nested.api));
+ assert_eq!(
+ nested.api.options().get(key),
+ dependency.api.options().get(key)
+ );
+ }
+
+ #[tokio::test]
+ async fn dependency_reads_preserve_an_explicit_read_via() {
+ let table = crate::table::rest_query_auth_table().await;
+ let mut env = table.rest_env().unwrap().clone();
+ let mut options = env.api.options().clone();
+ options.set(
+ "header.X-Paimon-Read-Via",
+ "an-already-encoded-outermost-table",
+ );
+ env.api = Arc::new(RESTApi::new(options, false).await.unwrap());
+ let dependency = env.for_dependency_reads().await.unwrap();
+ assert!(Arc::ptr_eq(&env.api, &dependency.api));
+ assert_eq!(
+ dependency
+ .api
+ .options()
+ .get("header.X-Paimon-Read-Via")
+ .unwrap(),
+ "an-already-encoded-outermost-table"
+ );
+ }
+
#[tokio::test]
async fn test_rest_env_clones_catalog_local_cache() {
let cache_directory = tempfile::tempdir().unwrap();
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 71c3d708..183984bd 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -1192,36 +1192,29 @@ impl<'a> PaimonTableRead<'a> {
.intersection(&view_fields)
.cloned()
.collect();
- let resolve_predicate_views =
- resolve_views && core_options.blob_as_descriptor() &&
!predicate_views.is_empty();
+ let resolve_predicate_views = resolve_views &&
!predicate_views.is_empty();
let mut output_views = view_fields.clone();
if resolve_predicate_views {
output_views.retain(|field| !predicate_views.contains(field));
}
- // Descriptor mode still translates view references to upstream
- // descriptors. Query predicates must see those logical values.
+ // User predicates see resolved descriptors or payloads, after
authorization.
let stream = match self.table.rest_env().filter(|_|
resolve_predicate_views) {
Some(env) =>
super::data_evolution_reader::resolve_blob_view_stream(
stream,
predicate_views,
env.clone(),
+ core_options.blob_as_descriptor(),
+ self.blob_parallelism,
),
None => stream,
};
- let mut payload_options = self.table.schema().options().clone();
- if resolve_views {
- let mut descriptors = core_options.blob_descriptor_fields();
- descriptors.extend(view_fields.iter().cloned());
- let mut descriptors: Vec<_> = descriptors.into_iter().collect();
- descriptors.sort_unstable();
- payload_options.insert("blob-descriptor-field".to_string(),
descriptors.join(","));
- }
- let payload_options = CoreOptions::new(&payload_options);
+ // Resolving views owns their upstream FileIO, including REST tokens.
+ // The ordinary Blob plan must not decode those payloads a second time.
let blob_plan = ManagedBlobReadPlan::new(
&self.read_type,
&self.data_predicates,
self.table.schema().fields(),
- &payload_options,
+ core_options,
);
let stream = match &blob_plan {
Some(plan) => {
@@ -1276,13 +1269,15 @@ impl<'a> PaimonTableRead<'a> {
stream,
output_views,
env.clone(),
+ core_options.blob_as_descriptor(),
+ self.blob_parallelism,
),
None => stream,
};
Ok(match blob_plan {
Some(plan) => plan.finish(
stream,
- &payload_options,
+ core_options,
self.table.file_io.clone(),
self.blob_parallelism,
self.limit,
@@ -1290,7 +1285,7 @@ impl<'a> PaimonTableRead<'a> {
None => resolve_primary_key_blob_stream(
stream,
&self.read_type,
- &payload_options,
+ core_options,
self.table.file_io.clone(),
self.blob_parallelism,
None,
diff --git a/crates/paimon/tests/mock_server.rs
b/crates/paimon/tests/mock_server.rs
index fd2082ac..ed127c0d 100644
--- a/crates/paimon/tests/mock_server.rs
+++ b/crates/paimon/tests/mock_server.rs
@@ -22,7 +22,7 @@
use axum::{
extract::{Extension, Json, Path, Query},
- http::StatusCode,
+ http::{HeaderMap, StatusCode},
response::IntoResponse,
routing::{get, post},
serve, Router,
@@ -70,6 +70,7 @@ struct MockState {
list_page_size: Option<usize>,
no_permission_databases: HashSet<String>,
no_permission_tables: HashSet<String>,
+ table_read_via: HashMap<String, String>,
create_partitions_calls: Vec<(String, String, CreatePartitionsRequest)>,
drop_partitions_calls: Vec<(String, String, DropPartitionsRequest)>,
create_partitions_error_status: Option<StatusCode>,
@@ -817,12 +818,31 @@ impl RESTServer {
/// Handle GET /databases/:db/tables/:table - get a specific table.
pub async fn get_table(
+ headers: HeaderMap,
Path((db, table)): Path<(String, String)>,
Extension(state): Extension<Arc<RESTServer>>,
) -> impl IntoResponse {
let mut s = state.inner.lock().unwrap();
let key = format!("{db}.{table}");
+ if let Some(expected) = s.table_read_via.get(&key) {
+ if headers
+ .get("X-Paimon-Read-Via")
+ .and_then(|value| value.to_str().ok())
+ != Some(expected)
+ {
+ return (
+ StatusCode::FORBIDDEN,
+ Json(ErrorResponse::new(
+ Some("table".to_string()),
+ Some(table.clone()),
+ Some("Dependency requires Read-Via".to_string()),
+ Some(403),
+ )),
+ )
+ .into_response();
+ }
+ }
if s.no_permission_tables.contains(&key) {
let err = ErrorResponse::new(
Some("table".to_string()),
@@ -1872,6 +1892,15 @@ impl RESTServer {
.create_partitions_statistics_error_status = status;
}
+ /// Require the dependency identity on get-table requests for this table.
+ pub fn require_table_read_via(&self, database: &str, table: &str,
read_via: String) {
+ self.inner
+ .lock()
+ .unwrap()
+ .table_read_via
+ .insert(format!("{database}.{table}"), read_via);
+ }
+
/// Make the list-partitions endpoint return the given status.
pub fn set_list_partitions_error_status(&self, status: Option<StatusCode>)
{
self.inner.lock().unwrap().list_partitions_error_status = status;
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 927f4136..36c1a35f 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -905,6 +905,19 @@ async fn
test_blob_view_limit_only_resolves_selected_references() {
.await
.unwrap();
+ // This is payload, not another reference to decode. Both the ordinary
+ // and authorized view pipelines below must return these bytes unchanged.
+ let first_payload =
+
paimon::spec::BlobDescriptor::new("memory:/not-an-upstream-payload".to_string(),
0, 4)
+ .serialize();
+ let payload_path = tmp.path().join("descriptor-payload");
+ std::fs::write(&payload_path, &first_payload).unwrap();
+ let first_input = paimon::spec::BlobDescriptor::new(
+ format!("file://{}", payload_path.display()),
+ 0,
+ first_payload.len() as i64,
+ )
+ .serialize();
let source_id = Identifier::new("default", "blob_source");
let source_schema = blob_schema(&[]);
fs_catalog
@@ -917,7 +930,7 @@ async fn
test_blob_view_limit_only_resolves_selected_references() {
blob_batch(
vec![1, 2],
vec!["Alice", "Bob"],
- vec![b"alice".to_vec(), b"bob".to_vec()],
+ vec![first_input, b"bob".to_vec()],
),
"source-writer",
)
@@ -983,6 +996,8 @@ async fn
test_blob_view_limit_only_resolves_selected_references() {
.await
.expect("create rest catalog");
+ let read_via =
paimon::api::RESTUtil::encode_string(&serde_json::to_string(&view_id).unwrap());
+ server.require_table_read_via("default", "blob_source", read_via);
let rest_view = rest_catalog.get_table(&view_id).await.unwrap();
let predicate = PredicateBuilder::new(rest_view.schema().fields())
.equal("id", Datum::Int(1))
@@ -1114,7 +1129,7 @@ async fn
test_blob_view_limit_only_resolves_selected_references() {
assert_eq!(
collect_blob_rows(&batches),
vec![
- (1, "Updated".to_string(), Some(b"alice".to_vec())),
+ (1, "Updated".to_string(), Some(first_payload.clone())),
(2, "Repeated".to_string(), None),
(3, "Repeated again".to_string(), Some(b"bob".to_vec())),
]
@@ -1140,7 +1155,7 @@ async fn
test_blob_view_limit_only_resolves_selected_references() {
assert_eq!(
collect_blob_rows(&read_all(&authorized.new_read_builder()).await.unwrap()),
vec![
- (1, "Updated".to_string(), Some(b"alice".to_vec())),
+ (1, "Updated".to_string(), Some(first_payload.clone())),
(2, "Repeated".to_string(), None),
(3, "Repeated again".to_string(), Some(b"bob".to_vec())),
]
@@ -1154,7 +1169,7 @@ async fn
test_blob_view_limit_only_resolves_selected_references() {
let mut filtered = authorized.new_read_builder();
filtered.with_projection(&["id"]).unwrap().with_filter(
PredicateBuilder::new(authorized.schema().fields())
- .equal("picture", Datum::Bytes(b"alice".to_vec()))
+ .equal("picture", Datum::Bytes(first_payload.clone()))
.unwrap(),
);
let rows: Vec<RecordBatch> = filtered
