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 1cb01974 feat(auth): support query-auth BLOB column masking (#1089)
1cb01974 is described below
commit 1cb01974c1f905c9b66a3a3e4160d355abc95f2f
Author: Jiajia Li <[email protected]>
AuthorDate: Sat Oct 10 15:13:07 2026 +0800
feat(auth): support query-auth BLOB column masking (#1089)
---
crates/paimon/src/table/audit_log_table/read.rs | 13 +-
crates/paimon/src/table/query_auth.rs | 2 +-
crates/paimon/src/table/query_auth/rules.rs | 28 --
crates/paimon/src/table/table_read.rs | 59 ++-
crates/paimon/tests/rest_catalog_test.rs | 555 +++++++++++++++++++++++-
docs/src/sql.md | 3 +-
6 files changed, 608 insertions(+), 52 deletions(-)
diff --git a/crates/paimon/src/table/audit_log_table/read.rs
b/crates/paimon/src/table/audit_log_table/read.rs
index f63b9c00..6efce91a 100644
--- a/crates/paimon/src/table/audit_log_table/read.rs
+++ b/crates/paimon/src/table/audit_log_table/read.rs
@@ -58,8 +58,16 @@ impl<'a> AuditLogRead<'a> {
/// Reads splits planned by an audit scan, retaining winning retract rows.
pub fn to_arrow(&self, data_splits: &[DataSplit]) ->
crate::Result<ArrowRecordBatchStream> {
// The primary-key path below builds its readers directly, so decide
here.
- self.read
-
.ensure_authorized_by_splits(&self.read.table.schema.core_options(),
data_splits)?;
+ let restricted = self
+ .read
+
.ensure_authorized_by_splits(&self.read.table.schema.core_options(),
data_splits)?
+ .is_some();
+ let has_primary_keys =
!self.read.table.schema().primary_keys().is_empty();
+ if restricted && has_primary_keys {
+ return Err(crate::table::query_auth::unsupported(
+ "a primary-key audit log read cannot apply a row filter or
column masking",
+ ));
+ }
let output_read_type = self.read.read_type.clone();
if output_read_type
.iter()
@@ -72,7 +80,6 @@ impl<'a> AuditLogRead<'a> {
});
}
let audit_schema = build_target_arrow_schema(&output_read_type)?;
- let has_primary_keys =
!self.read.table.schema().primary_keys().is_empty();
let mut read_type: Vec<_> = output_read_type
.iter()
.filter(|field| field.id() != ROW_KIND_FIELD_ID)
diff --git a/crates/paimon/src/table/query_auth.rs
b/crates/paimon/src/table/query_auth.rs
index 06625d9d..3ae7d349 100644
--- a/crates/paimon/src/table/query_auth.rs
+++ b/crates/paimon/src/table/query_auth.rs
@@ -20,7 +20,7 @@
mod rules;
use crate::api::AuthTableQueryResponse;
-pub(crate) use rules::{filter_batch, mask_batch, mask_inputs, Rules};
+pub(crate) use rules::{filter_batch, mask_batch, mask_inputs, ColumnMask,
Rules};
/// The server's answer for one user on one table; `session` ties it to the
/// handle that asked, as the response names no table or user.
diff --git a/crates/paimon/src/table/query_auth/rules.rs
b/crates/paimon/src/table/query_auth/rules.rs
index 66dc89f3..08650011 100644
--- a/crates/paimon/src/table/query_auth/rules.rs
+++ b/crates/paimon/src/table/query_auth/rules.rs
@@ -86,9 +86,6 @@ impl Rules {
"the server masks '{column}', which is not a column"
))
})?;
- if fields[target].data_type().is_blob_file_field() {
- return Err(unsupported("BLOB column masks are not supported
yet"));
- }
let transform = Transform::from_rest_json(json,
fields).map_err(|e| {
unsupported(&format!(
"cannot parse the server's mask on '{column}': {e}"
@@ -814,31 +811,6 @@ mod tests {
assert!(refusal(parse(&[&good, ""], &[], &fields)).contains("empty row
filter"));
}
- #[test]
- fn test_blob_masks_are_not_supported_yet() {
- use crate::spec::{ArrayType, BlobType, MapType};
-
- let blob = DataType::Blob(BlobType::new());
- for data_type in [
- blob.clone(),
- DataType::Array(ArrayType::new(blob.clone())),
- DataType::Map(MapType::new(string(), blob)),
- ] {
- let fields = vec![field(0, "payload", data_type.clone())];
- for mask in [
- NULL.to_string(),
- field_ref("payload"),
- cast("payload", &data_type.to_string()),
- ] {
- let message = refusal(parse(&[], &[("payload", &mask)],
&fields));
- assert!(
- message.contains("BLOB column masks are not supported
yet"),
- "{message}"
- );
- }
- }
- }
-
#[test]
fn test_parse_refuses_a_bad_mask() {
let fields = rule_fields();
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 94c1ae8b..1b0e2587 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -1191,6 +1191,7 @@ impl<'a> PaimonTableRead<'a> {
row_filter_factory: None,
file_fields: schema_fields.clone(),
};
+ let blob_options = has_blobs.then(|| self.masked_blob_options(&masks));
let stream = stream.map(move |batch| {
let batch = batch?;
let names_match = batch.num_columns() == physical.len()
@@ -1208,6 +1209,7 @@ impl<'a> PaimonTableRead<'a> {
let batch = filter_batch(&batch, &filters, &schema_fields,
&physical)?;
// Safe ordinary conjuncts avoid masking rows the caller excludes.
let batch = filter_record_batch_by_predicates(batch, &predicates,
&physical)?;
+ // BLOB masks copy references; consumers resolve payloads lazily.
let batch = mask_batch(&batch, &masks, &schema_fields, &physical)?;
if has_blobs {
Ok(batch)
@@ -1216,8 +1218,8 @@ impl<'a> PaimonTableRead<'a> {
}
});
let stream: ArrowRecordBatchStream = Box::pin(stream);
- if has_blobs {
- return self.finish_authorized_blobs(stream, &batch_fields,
&pending, &core_options);
+ if let Some(options) = blob_options {
+ return self.finish_authorized_blobs(stream, &batch_fields,
&pending, options);
}
let stream = project_authorized_stream(stream, self.read_type.len());
Ok(match self.limit {
@@ -1226,15 +1228,66 @@ impl<'a> PaimonTableRead<'a> {
})
}
+ /// Copy the source's BLOB representation along with its references.
+ fn masked_blob_options(
+ &self,
+ masks: &[super::query_auth::ColumnMask],
+ ) -> std::collections::HashMap<String, String> {
+ use crate::spec::Transform;
+
+ let fields = self.table.schema().fields();
+ let copies: Vec<_> = masks
+ .iter()
+ .filter_map(|mask| {
+ let source = match &mask.transform {
+ Transform::FieldRef(index) | Transform::Cast(index, _) =>
&fields[*index],
+ _ => return None,
+ };
+ let target = &fields[mask.column];
+ (source.data_type().is_blob_file_field()
+ && target.data_type().is_blob_file_field()
+ && source.name() != target.name())
+ .then(|| (source.name(), target.name()))
+ })
+ .collect();
+ let mut options = self.table.schema().options().clone();
+ if copies.is_empty() {
+ return options;
+ }
+ let core_options = self.table.schema().core_options();
+ for (key, original) in [
+ (
+ "blob-descriptor-field",
+ core_options.blob_descriptor_fields(),
+ ),
+ ("blob-view-field", core_options.blob_view_fields()),
+ ("video-frame-field", core_options.video_frame_fields()),
+ ] {
+ let mut selected = original.clone();
+ for &(source, target) in &copies {
+ if original.contains(source) {
+ selected.insert(target.to_string());
+ } else {
+ selected.remove(target);
+ }
+ }
+ let mut selected: Vec<_> = selected.into_iter().collect();
+ selected.sort_unstable();
+ options.insert(key.to_string(), selected.join(","));
+ }
+ options
+ }
+
fn finish_authorized_blobs(
&self,
stream: ArrowRecordBatchStream,
batch_fields: &[DataField],
predicates: &[Predicate],
- core_options: &CoreOptions<'_>,
+ payload_options: std::collections::HashMap<String, String>,
) -> crate::Result<ArrowRecordBatchStream> {
use super::managed_blob_reader::{resolve_primary_key_blob_stream,
ManagedBlobReadPlan};
+ let core_options = &CoreOptions::new(&payload_options);
let view_fields = core_options.blob_view_fields();
let resolve_views = core_options.blob_view_resolve_enabled()
&& self.table.rest_env().is_some()
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index c6882fb5..c95b91a4 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -1225,6 +1225,40 @@ async fn
test_blob_view_limit_only_resolves_selected_references() {
);
assert_eq!(ids(&read_all(&filtered).await.unwrap()), vec![1]);
+ // Masked views resolve once when their values are read.
+ for (mask, expected) in [
+ (serde_json::json!({"name": "NULL"}), vec![None, None, None]),
+ (
+ field_ref_mask(2, "picture", "BLOB"),
+ vec![Some(first_payload.clone()), None, Some(b"bob".to_vec())],
+ ),
+ ] {
+ server.set_auth_response(
+ "default",
+ "blob_view_target",
+ rules(&[int_leaf(0, "id", "LESS_THAN", 4)], &[("picture", mask)]),
+ );
+ let rows =
collect_blob_rows(&read_all(&authorized.new_read_builder()).await.unwrap());
+ assert_eq!(
+ rows.into_iter().map(|row| row.2).collect::<Vec<_>>(),
+ expected
+ );
+ let mut filtered = authorized.new_read_builder();
+ filtered.with_projection(&["id"]).unwrap().with_filter(
+ PredicateBuilder::new(authorized.schema().fields())
+ .equal("picture", Datum::Bytes(first_payload.clone()))
+ .unwrap(),
+ );
+ assert_eq!(
+ ids(&read_all(&filtered).await.unwrap()),
+ if expected[0].is_some() {
+ vec![1]
+ } else {
+ vec![]
+ },
+ );
+ }
+
// A view reference and its upstream BLOB can disagree on nullness.
// Until rules can resolve that dependency safely, reject instead of
// authorizing rows by the reference bytes.
@@ -3045,36 +3079,505 @@ async fn
test_query_auth_filters_before_resolving_blob_payloads() {
.await
.unwrap();
assert_eq!(ids(&result), vec![3]);
+
+ // BLOB masks still return logical payloads.
+ g.set_rules(
+ &[int_leaf(0, "id", "NOT_EQUAL", 2)],
+ &[("payload", field_ref_mask(1, "payload", "BLOB"))],
+ );
+ let batches = read_all(&g.table.new_read_builder()).await.unwrap();
+ assert_eq!(ids(&batches), vec![1, 3]);
+ assert_eq!(
+ batches
+ .iter()
+ .flat_map(|batch| column::<LargeBinaryArray>(batch,
"payload").iter())
+ .flatten()
+ .collect::<Vec<_>>(),
+ vec![b"good".as_slice()],
+ );
+ assert_eq!(ids(&read_all(&filtered).await.unwrap()), vec![3]);
+
+ // A NULL mask must not open the missing payload.
+ g.set_rules(
+ &[int_leaf(0, "id", "GREATER_THAN", 1)],
+ &[("payload", serde_json::json!({"name": "NULL"}))],
+ );
+ let batches = read_all(&g.table.new_read_builder()).await.unwrap();
+ assert_eq!(ids(&batches), vec![2, 3]);
+ for batch in &batches {
+ assert_eq!(
+ column::<LargeBinaryArray>(batch, "payload").null_count(),
+ batch.num_rows(),
+ );
+ }
+ assert!(read_all(&filtered)
+ .await
+ .unwrap()
+ .iter()
+ .all(|b| b.num_rows() == 0));
+ let mut nulls = g.table.new_read_builder();
+ nulls.with_projection(&["id"]).unwrap().with_filter(
+ PredicateBuilder::new(g.table.schema().fields())
+ .is_null("payload")
+ .unwrap(),
+ );
+ assert_eq!(ids(&read_all(&nulls).await.unwrap()), vec![2, 3]);
+ nulls.with_limit(1);
+ assert_eq!(ids(&read_all(&nulls).await.unwrap()), vec![2]);
+ }
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_blob_masks_preserve_filter_and_limit_selection() {
+ let (_source, good, missing) = blob_payloads();
+ for primary_key in [true, false] {
+ let g = blob_mask_rows(primary_key, vec![None, Some(&missing),
Some(&good)]).await;
+ for masked in [false, true] {
+ let masks = if masked {
+ vec![
+ ("payload", field_ref_mask(1, "payload", "BLOB")),
+ ("label", upper(2, "label")),
+ ]
+ } else {
+ vec![]
+ };
+ g.set_rules(&[int_leaf(0, "id", "GREATER_THAN", 0)], &masks);
+ for batch_size in [1, 3] {
+ let table = g.table.copy_with_options(HashMap::from([(
+ "read.batch-size".to_string(),
+ batch_size.to_string(),
+ )]));
+ let predicates =
PredicateBuilder::new(table.schema().fields());
+ for (case, predicate, limit, expected) in [
+ (
+ "ordinary filter",
+ Some(predicates.equal("id", Datum::Int(3)).unwrap()),
+ None,
+ vec![3],
+ ),
+ (
+ "scalar mask filter",
+ Some(
+ predicates
+ .equal(
+ "label",
+ Datum::String(if masked { "ROW3" } else {
"row3" }.to_string()),
+ )
+ .unwrap(),
+ ),
+ None,
+ vec![3],
+ ),
+ ("limit", None, Some(1), vec![1]),
+ (
+ "remaining predicate",
+ Some(paimon::spec::Predicate::and(vec![
+ predicates.not_equal("id", Datum::Int(2)).unwrap(),
+ predicates
+ .equal("payload",
Datum::Bytes(b"good".to_vec()))
+ .unwrap(),
+ ])),
+ Some(1),
+ vec![3],
+ ),
+ (
+ "or predicate",
+ Some(paimon::spec::Predicate::or(vec![
+ predicates.equal("id", Datum::Int(3)).unwrap(),
+ predicates.is_null("payload").unwrap(),
+ ])),
+ Some(1),
+ vec![1],
+ ),
+ ("zero limit", None, Some(0), vec![]),
+ ] {
+ let mut read = table.new_read_builder();
+ if let Some(predicate) = predicate {
+ read.with_filter(predicate);
+ }
+ if let Some(limit) = limit {
+ read.with_limit(limit);
+ }
+ let label =
+ format!("pk={primary_key}, mask={masked},
batch={batch_size}, {case}");
+ let batches = read_all(&read)
+ .await
+ .unwrap_or_else(|error| panic!("{label}: {error}"));
+ assert_eq!(ids(&batches), expected, "{label}");
+ for batch in &batches {
+ let id = column::<Int32Array>(batch, "id");
+ let payload = column::<LargeBinaryArray>(batch,
"payload");
+ for row in 0..batch.num_rows() {
+ if id.value(row) == 1 {
+ assert!(payload.is_null(row), "{label}");
+ } else {
+ assert_eq!(payload.value(row), b"good",
"{label}");
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_blob_masks_keep_predicates_and_projection_lazy() {
+ let (_source, good, missing) = blob_payloads();
+ let field = serde_json::json!({"index": 1, "name": "payload", "type":
"BLOB"});
+ for primary_key in [true, false] {
+ let g = blob_mask_rows(primary_key, vec![Some(&missing), None,
Some(&good)]).await;
+ for (name, mask) in [
+ ("unmasked", None),
+ (
+ "field-ref",
+ Some(serde_json::json!({"name": "FIELD_REF", "fieldRef":
field})),
+ ),
+ (
+ "cast",
+ Some(serde_json::json!({"name": "CAST", "fieldRef": field,
"type": "BLOB"})),
+ ),
+ ] {
+ let masks: Vec<_> = mask.into_iter().map(|mask| ("payload",
mask)).collect();
+ g.set_rules(&[int_leaf(0, "id", "GREATER_THAN", 0)], &masks);
+ for batch_size in [1, 3] {
+ let table = g.table.copy_with_options(HashMap::from([(
+ "read.batch-size".to_string(),
+ batch_size.to_string(),
+ )]));
+ let predicates =
PredicateBuilder::new(table.schema().fields());
+ let first = predicates.equal("id", Datum::Int(1)).unwrap();
+ let payload = predicates
+ .equal("payload", Datum::Bytes(b"good".to_vec()))
+ .unwrap();
+ for (case, predicate, expected) in [
+ (
+ "or",
+ paimon::spec::Predicate::or(vec![first.clone(),
payload.clone()]),
+ vec![1, 3],
+ ),
+ (
+ "is-not-null",
+ predicates.is_not_null("payload").unwrap(),
+ vec![1, 3],
+ ),
+ ("is-null", predicates.is_null("payload").unwrap(),
vec![2]),
+ (
+ "nested-or",
+ paimon::spec::Predicate::or(vec![
+ paimon::spec::Predicate::and(vec![
+ first.clone(),
+ predicates.is_not_null("payload").unwrap(),
+ ]),
+ payload.clone(),
+ ]),
+ vec![1, 3],
+ ),
+ ] {
+ for limit in [None, Some(1)] {
+ let mut read = table.new_read_builder();
+ read.with_projection(&["id"])
+ .unwrap()
+ .with_filter(predicate.clone());
+ if let Some(limit) = limit {
+ read.with_limit(limit);
+ }
+ let label = format!(
+ "pk={primary_key}, {name}, batch={batch_size},
{case}, limit={limit:?}"
+ );
+ let batches = read_all(&read)
+ .await
+ .unwrap_or_else(|error| panic!("{label}:
{error}"));
+ assert_eq!(
+ ids(&batches),
+ expected[..limit.unwrap_or(expected.len())],
+ "{label}"
+ );
+ }
+ }
+ for (projection, predicate) in [(vec!["payload"], first),
(vec!["id"], payload)] {
+ let mut read = table.new_read_builder();
+ read.with_projection(&projection)
+ .unwrap()
+ .with_filter(predicate);
+ let error = read_all(&read).await.unwrap_err().to_string();
+ assert!(error.contains("missing.blob"), "{error}");
+ }
+ let mut read = table.new_read_builder();
+ read.with_projection(&["payload"])
+ .unwrap()
+ .with_filter(predicates.equal("id",
Datum::Int(3)).unwrap());
+ let batches = read_all(&read).await.unwrap();
+
assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ let batch = batches.iter().find(|batch| batch.num_rows() >
0).unwrap();
+ assert_eq!(
+ column::<LargeBinaryArray>(batch, "payload").value(0),
+ b"good"
+ );
+ }
+ }
}
}
#[cfg(not(windows))]
#[tokio::test]
-async fn test_query_auth_refuses_blob_masks() {
+async fn test_blob_mask_alias_keeps_the_source_reference_format() {
+ let (_source, good, missing) = blob_payloads();
for primary_key in [true, false] {
- let g = blob_mask_rows(primary_key, vec![None]).await;
+ let batch = RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![1, 2, 3])) as
arrow_array::ArrayRef,
+ ),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from(vec![
+ Some(missing.as_slice()),
+ None,
+ Some(good.as_slice()),
+ ])) as arrow_array::ArrayRef,
+ ),
+ (
+ "alias",
+ Arc::new(LargeBinaryArray::from(vec![None::<&[u8]>; 3])) as
arrow_array::ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let g = written(
+ "auth_blob_alias",
+ |options| {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::with_nullable(false)))
+ .column("payload", DataType::Blob(BlobType::new()))
+ .column("alias", DataType::Blob(BlobType::new()))
+ .option("blob-descriptor-field", "payload")
+ .option("blob-view-field", "alias");
+ if primary_key {
+ schema = schema.primary_key(["id"]).option("bucket", "1");
+ } else {
+ schema = schema
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ }
+ for (key, value) in options {
+ schema = schema.option(*key, *value);
+ }
+ schema.build().unwrap()
+ },
+ vec![batch],
+ )
+ .await;
+ let field = serde_json::json!({"index": 1, "name": "payload", "type":
"BLOB"});
for mask in [
- serde_json::json!({"name": "NULL"}),
- field_ref_mask(1, "payload", "BLOB"),
- serde_json::json!({"name": "CAST", "fieldRef": {
- "index": 1, "name": "payload", "type": "BLOB",
- }, "type": "BLOB"}),
+ serde_json::json!({"name": "FIELD_REF", "fieldRef": field}),
+ serde_json::json!({"name": "CAST", "fieldRef": field, "type":
"BLOB"}),
] {
- g.set_rules(&[], &[("payload", mask)]);
- for projection in [vec!["payload"], vec!["id"], vec![]] {
+ g.set_rules(&[int_leaf(0, "id", "GREATER_THAN", 0)], &[("alias",
mask)]);
+ let predicates = PredicateBuilder::new(g.table.schema().fields());
+ for (predicate, expected) in [
+ (
+ paimon::spec::Predicate::or(vec![
+ predicates.equal("id", Datum::Int(1)).unwrap(),
+ predicates
+ .equal("alias", Datum::Bytes(b"good".to_vec()))
+ .unwrap(),
+ ]),
+ vec![1, 3],
+ ),
+ (predicates.is_not_null("alias").unwrap(), vec![1, 3]),
+ (predicates.is_null("alias").unwrap(), vec![2]),
+ ] {
let mut read = g.table.new_read_builder();
- read.with_projection(&projection).unwrap();
- let error = read.new_scan().plan().await.unwrap_err();
- assert!(
- matches!(error, paimon::Error::Unsupported { ref message }
- if message.contains("BLOB column masks are not supported
yet")),
- "{error:?}"
- );
+ read.with_projection(&["id"])
+ .unwrap()
+ .with_filter(predicate);
+ assert_eq!(ids(&read_all(&read).await.unwrap()), expected);
}
+ let mut read = g.table.new_read_builder();
+ read.with_projection(&["alias"])
+ .unwrap()
+ .with_filter(predicates.equal("id", Datum::Int(3)).unwrap());
+ let batches = read_all(&read).await.unwrap();
+ let batch = batches.iter().find(|batch| batch.num_rows() >
0).unwrap();
+ assert_eq!(column::<LargeBinaryArray>(batch, "alias").value(0),
b"good");
+ read.with_filter(predicates.equal("id", Datum::Int(1)).unwrap());
+ let error = read_all(&read).await.unwrap_err().to_string();
+ assert!(error.contains("missing.blob"), "{error}");
}
}
}
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_blob_view_mask_alias_resolves_the_source_view() {
+ 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).unwrap();
+ fs_catalog
+ .create_database("default", true, HashMap::new())
+ .await
+ .unwrap();
+
+ let source_id = Identifier::new("default", "alias_view_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 picture_field_id = source
+ .schema()
+ .fields()
+ .iter()
+ .find(|field| field.name() == "picture")
+ .unwrap()
+ .id();
+ let view_ref = |row_id| {
+ BlobViewStruct::new(source_id.clone(), picture_field_id, row_id)
+ .serialize()
+ .unwrap()
+ };
+
+ // Only `picture` is a view; the mask makes `alias` one by copying it.
+ let view_id = Identifier::new("default", "alias_view_target");
+ let view_schema = |options: &[(&str, &str)]| {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("picture", DataType::Blob(BlobType::new()))
+ .column("alias", DataType::Blob(BlobType::new()))
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .option("blob-view-field", "picture");
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().unwrap()
+ };
+ fs_catalog
+ .create_table(&view_id, view_schema(&[]), false)
+ .await
+ .unwrap();
+ let view = fs_catalog.get_table(&view_id).await.unwrap();
+ let (first, second) = (view_ref(0), view_ref(1));
+ let batch = RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![1, 2, 3])) as arrow_array::ArrayRef,
+ ),
+ (
+ "picture",
+ Arc::new(LargeBinaryArray::from(vec![
+ Some(first.as_slice()),
+ None,
+ Some(second.as_slice()),
+ ])) as arrow_array::ArrayRef,
+ ),
+ (
+ "alias",
+ Arc::new(LargeBinaryArray::from(vec![None::<&[u8]>; 3])) as
arrow_array::ArrayRef,
+ ),
+ ])
+ .unwrap();
+ write_batch(&view, batch, "view-writer").await;
+
+ let mut defaults = HashMap::new();
+ defaults.insert("prefix".to_string(), "mock-test".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",
+ "alias_view_source",
+ source_schema,
+ source.location(),
+ );
+ server.add_table_with_schema(
+ "default",
+ "alias_view_target",
+ view_schema(&[("query-auth.enabled", "true")]),
+ view.location(),
+ );
+ let read_via =
paimon::api::RESTUtil::encode_string(&serde_json::to_string(&view_id).unwrap());
+ server.require_table_read_via("default", "alias_view_source", read_via);
+ server.set_auth_response(
+ "default",
+ "alias_view_target",
+ rules(&[], &[("alias", field_ref_mask(1, "picture", "BLOB"))]),
+ );
+ let mut rest_options = Options::new();
+ rest_options.set("uri", server.url().unwrap());
+ 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.unwrap();
+ let table = rest_catalog.get_table(&view_id).await.unwrap();
+
+ let payloads = |batches: &[RecordBatch], name: &str| {
+ let mut rows: Vec<(i32, Option<Vec<u8>>)> = batches
+ .iter()
+ .flat_map(|b| {
+ let ids = column::<Int32Array>(b, "id").values().to_vec();
+ let values = column::<LargeBinaryArray>(b, name).iter();
+ ids.into_iter().zip(values.map(|v| v.map(|v| v.to_vec())))
+ })
+ .collect();
+ rows.sort();
+ rows
+ };
+ let expected = vec![
+ (1, Some(b"alice".to_vec())),
+ (2, None),
+ (3, Some(b"bob".to_vec())),
+ ];
+ // The alias resolves to the source view's payload, alone or beside it.
+ for projection in [vec!["id", "alias"], vec!["id", "picture", "alias"]] {
+ let mut read = table.new_read_builder();
+ read.with_projection(&projection).unwrap();
+ let batches = read_all(&read).await.unwrap();
+ assert_eq!(payloads(&batches, "alias"), expected);
+ if projection.contains(&"picture") {
+ assert_eq!(payloads(&batches, "picture"), expected);
+ }
+ }
+
+ // Predicates on the alias match the resolved payload, not the reference.
+ let predicates = PredicateBuilder::new(table.schema().fields());
+ for (predicate, expected) in [
+ (
+ predicates
+ .equal("alias", Datum::Bytes(b"bob".to_vec()))
+ .unwrap(),
+ vec![3],
+ ),
+ (predicates.is_null("alias").unwrap(), vec![2]),
+ (predicates.is_not_null("alias").unwrap(), vec![1, 3]),
+ ] {
+ let mut read = table.new_read_builder();
+ read.with_projection(&["id"])
+ .unwrap()
+ .with_filter(predicate);
+ assert_eq!(ids(&read_all(&read).await.unwrap()), expected);
+ }
+}
+
fn schema_of(columns: &[&str], options: &[(&str, &str)]) -> Schema {
let mut builder = Schema::builder();
for name in columns {
@@ -4623,6 +5126,26 @@ async fn
test_query_auth_masks_the_merged_primary_key_row() {
);
}
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_query_auth_refuses_a_restricted_primary_key_audit_log_read() {
+ let g = scores().await;
+ g.set_rules(
+ &[int_leaf(2, "score", "GREATER_THAN", 30)],
+ &[("name", upper(1, "name"))],
+ );
+ // An audit scan is refused at planning, but ordinary splits still carry
+ // the grant to a directly built audit read.
+ let builder = g.table.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let audit =
paimon::table::AuditLogRead::new(builder.new_read().unwrap()).unwrap();
+ // It builds its primary-key readers itself, so it must not return raw
rows.
+ let Err(err) = audit.to_arrow(plan.splits()) else {
+ panic!("a restricted primary-key audit read must be refused")
+ };
+ assert_refused(err);
+}
+
#[tokio::test]
async fn test_rest_catalog_manages_permissions_end_to_end() {
let ctx = setup_catalog(vec!["default"]).await;
diff --git a/docs/src/sql.md b/docs/src/sql.md
index b9a2e64f..6de93f0e 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1650,7 +1650,8 @@ query asks the catalog what the current user may read,
and the row filters and
column masks it returns (see `create_policy`) are applied to the result:
- Rows are filtered on their stored values; masked columns return masked
values.
-- BLOB column masks are not yet supported.
+- BLOB masks keep payload reads lazy. A NULL mask does not read the original
+ BLOB payload.
- Query filters on masked columns, including partition keys, match masked
values.
- Statistics are not used under a row filter or a mask, and a row filter turns
off `LIMIT` pushdown, so such queries read the data.
