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 637386b1 feat(auth): apply query-auth masks to non-BLOB columns (#1077)
637386b1 is described below
commit 637386b196da0b017557a1fdc01c518834732fb9
Author: Jiajia Li <[email protected]>
AuthorDate: Sat Oct 10 11:57:39 2026 +0800
feat(auth): apply query-auth masks to non-BLOB columns (#1077)
---
.../datafusion/src/physical_plan/scan.rs | 2 +-
crates/integrations/datafusion/tests/query_auth.rs | 145 +++-
crates/paimon/src/table/query_auth.rs | 2 +-
crates/paimon/src/table/query_auth/rules.rs | 957 ++++++++++++++++++++-
crates/paimon/src/table/read_builder.rs | 17 +-
crates/paimon/src/table/source.rs | 5 +-
crates/paimon/src/table/table_read.rs | 88 +-
crates/paimon/src/table/table_scan.rs | 96 ++-
crates/paimon/tests/rest_catalog_test.rs | 863 +++++++++++++++++--
docs/src/sql.md | 15 +-
10 files changed, 2043 insertions(+), 147 deletions(-)
diff --git a/crates/integrations/datafusion/src/physical_plan/scan.rs
b/crates/integrations/datafusion/src/physical_plan/scan.rs
index 4d47ccb1..0a9eb5ab 100644
--- a/crates/integrations/datafusion/src/physical_plan/scan.rs
+++ b/crates/integrations/datafusion/src/physical_plan/scan.rs
@@ -1193,7 +1193,7 @@ impl ExecutionPlan for PaimonTableScan {
}
fn partition_statistics(&self, partition: Option<usize>) ->
DFResult<Arc<Statistics>> {
- // They include the rows the server's rules drop.
+ // Statistics include hidden rows and unmasked values.
if self.has_query_auth_rules() {
return Ok(Arc::new(Statistics::new_unknown(&self.schema())));
}
diff --git a/crates/integrations/datafusion/tests/query_auth.rs
b/crates/integrations/datafusion/tests/query_auth.rs
index 2ef11688..7f821577 100644
--- a/crates/integrations/datafusion/tests/query_auth.rs
+++ b/crates/integrations/datafusion/tests/query_auth.rs
@@ -15,8 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-//! SQL over a `query-auth.enabled` table whose REST server sets a row filter
-//! for the current user.
+//! SQL row filtering and column masking through a REST catalog.
mod common;
@@ -54,6 +53,13 @@ fn schema(options: &[(&str, &str)]) -> Schema {
/// catalog that admits `id > 6` and applies `column_masking`.
async fn restricted_people(
column_masking: Option<HashMap<String, String>>,
+) -> (tempfile::TempDir, RESTServer, SQLContext) {
+ restricted_people_with_schema(column_masking, schema).await
+}
+
+async fn restricted_people_with_schema(
+ column_masking: Option<HashMap<String, String>>,
+ schema: impl Fn(&[(&str, &str)]) -> Schema,
) -> (tempfile::TempDir, RESTServer, SQLContext) {
let tmp = tempfile::tempdir().unwrap();
let mut fs_options = Options::new();
@@ -141,6 +147,18 @@ async fn restricted_people(
(tmp, server, context)
}
+/// Upper-cases `name` and hides `secret`.
+fn masks() -> Option<HashMap<String, String>> {
+ let name = json!({"index": 1, "name": "name", "type": "VARCHAR(255)"});
+ Some(HashMap::from([
+ (
+ "name".to_string(),
+ json!({"name": "UPPER", "inputs": [name]}).to_string(),
+ ),
+ ("secret".to_string(), json!({"name": "NULL"}).to_string()),
+ ]))
+}
+
async fn query(context: &SQLContext, sql: &str) -> Vec<RecordBatch> {
context.sql(sql).await.unwrap().collect().await.unwrap()
}
@@ -231,6 +249,129 @@ async fn
test_query_auth_where_runs_on_the_admitted_rows_only() {
);
}
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_query_auth_where_on_a_masked_column_matches_masked_values() {
+ let (_tmp, _server, context) = restricted_people(masks()).await;
+
+ assert_eq!(
+ id_names(&query(&context, &format!("SELECT id, name FROM
{TABLE}")).await),
+ vec![
+ (7, "GRACE".to_string()),
+ (8, "HEIDI".to_string()),
+ (9, "IVAN".to_string()),
+ (10, "JUDY".to_string()),
+ ]
+ );
+ assert_eq!(
+ id_names(
+ &query(
+ &context,
+ &format!("SELECT id, name FROM {TABLE} WHERE name = 'GRACE'")
+ )
+ .await
+ ),
+ vec![(7, "GRACE".to_string())]
+ );
+ assert!(id_names(
+ &query(
+ &context,
+ &format!("SELECT id, name FROM {TABLE} WHERE name = 'grace'")
+ )
+ .await
+ )
+ .is_empty());
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_query_auth_filters_masked_partition_values() {
+ let (_tmp, server, context) = restricted_people_with_schema(masks(),
|options| {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::new(255).unwrap()))
+ .column("secret",
DataType::VarChar(VarCharType::new(255).unwrap()))
+ .partition_keys(["name"]);
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().unwrap()
+ })
+ .await;
+ server.set_auth_response(
+ "default",
+ "people",
+ AuthTableQueryResponse {
+ filter: None,
+ column_masking: masks(),
+ },
+ );
+
+ assert_eq!(
+ id_names(
+ &query(
+ &context,
+ &format!("SELECT id, name FROM {TABLE} WHERE name = 'HEIDI'")
+ )
+ .await
+ ),
+ vec![(8, "HEIDI".to_string())],
+ );
+ assert!(id_names(
+ &query(
+ &context,
+ &format!("SELECT id, name FROM {TABLE} WHERE name = 'heidi'")
+ )
+ .await
+ )
+ .is_empty());
+ assert_eq!(
+ count(
+ &query(
+ &context,
+ &format!("SELECT COUNT(*) FROM {TABLE} WHERE name = 'HEIDI'")
+ )
+ .await
+ ),
+ 1,
+ );
+ let selected = single(
+ &query(
+ &context,
+ &format!("SELECT id FROM {TABLE} WHERE name = 'HEIDI' LIMIT 1"),
+ )
+ .await,
+ );
+ assert_eq!(
+ selected
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 8
+ );
+ assert_eq!(
+ id_names(
+ &query(
+ &context,
+ &format!("SELECT id, name FROM {TABLE} WHERE name = 'HEIDI' OR
id = 1")
+ )
+ .await
+ ),
+ vec![(1, "ALICE".to_string()), (8, "HEIDI".to_string())],
+ );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn
test_query_auth_aggregates_see_masked_values_not_the_manifest_statistics() {
+ let (_tmp, _server, context) = restricted_people(masks()).await;
+
+ assert_eq!(
+ count(&query(&context, &format!("SELECT COUNT(secret) FROM
{TABLE}")).await),
+ 0
+ );
+ let max = single(&query(&context, &format!("SELECT MAX(secret) FROM
{TABLE}")).await);
+ assert!(max.is_null(0), "{max:?}");
+}
+
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_query_auth_explain_shows_the_restriction_but_not_the_files() {
let (_tmp, _server, context) = restricted_people(None).await;
diff --git a/crates/paimon/src/table/query_auth.rs
b/crates/paimon/src/table/query_auth.rs
index 7614c602..06625d9d 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, Rules};
+pub(crate) use rules::{filter_batch, mask_batch, mask_inputs, 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 20710390..66dc89f3 100644
--- a/crates/paimon/src/table/query_auth/rules.rs
+++ b/crates/paimon/src/table/query_auth/rules.rs
@@ -15,22 +15,30 @@
// specific language governing permissions and limitations
// under the License.
-//! The server's row filter, parsed against the schema it ruled on and applied
-//! to read batches, as Java `TableQueryAuthResult` does.
+//! Apply the server's row filters and column masks to read batches.
use super::unsupported;
use crate::api::AuthTableQueryResponse;
use crate::arrow::residual::{evaluate_exact_leaf_predicate,
sanitize_filter_mask};
-use crate::spec::{DataField, Predicate};
+use crate::spec::{DataField, Predicate, Transform, TransformInput};
use crate::Result;
+use arrow_array::cast::AsArray;
use arrow_array::{Array, ArrayRef, BooleanArray, Float32Array, Float64Array,
RecordBatch};
use std::collections::HashSet;
use std::sync::Arc;
-/// The rules of one grant; leaf indices are table-schema positions.
+/// Masks one column, by table-schema index.
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) struct ColumnMask {
+ pub(crate) column: usize,
+ pub(crate) transform: Transform,
+}
+
+/// One grant's rules, indexed by table schema.
#[derive(Debug, Default, PartialEq)]
pub(crate) struct Rules {
pub(crate) filters: Vec<Predicate>,
+ pub(crate) masks: Vec<ColumnMask>,
}
impl Rules {
@@ -54,19 +62,71 @@ impl Rules {
filters.push(filter);
}
- // Not applied yet: refusing beats returning raw values.
- if response
- .column_masking
- .as_ref()
- .is_some_and(|masks| !masks.is_empty())
- {
- return Err(unsupported("this client does not apply column masking
yet"));
+ let mut masks = Vec::new();
+ for (column, json) in response.column_masking.iter().flatten() {
+ if column.is_empty() || json.is_empty() {
+ return Err(unsupported("the server sent an empty column
mask"));
+ }
+ // Unprojected system columns need no masking; `_KEY_` names are
not exempt.
+ if crate::spec::is_reserved_system_field_name(column) &&
!column.starts_with("_KEY_") {
+ // Still reject malformed masks.
+ let parsed = serde_json::from_str::<serde_json::Value>(json);
+ if !parsed.is_ok_and(|value| !value.is_null()) {
+ return Err(unsupported(&format!(
+ "cannot parse the server's mask on '{column}'"
+ )));
+ }
+ continue;
+ }
+ let target = fields
+ .iter()
+ .position(|f| f.name() == column)
+ .ok_or_else(|| {
+ unsupported(&format!(
+ "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}"
+ ))
+ })?;
+ check_mask_fits(&transform, &fields[target], fields)?;
+ masks.push(ColumnMask {
+ column: target,
+ transform,
+ });
}
- Ok(Self { filters })
+ masks.sort_by_key(|m| m.column);
+
+ // Reading another masked column would expose its raw value.
+ let targets: HashSet<usize> = masks.iter().map(|m| m.column).collect();
+ for mask in &masks {
+ if let Some(other) = mask_inputs(mask)
+ .into_iter()
+ .find(|i| *i != mask.column && targets.contains(i))
+ {
+ return Err(unsupported(&format!(
+ "the mask on '{}' reads '{}', which is masked too, so it
would expose the \
+ raw value",
+ fields[mask.column].name(),
+ fields[other].name()
+ )));
+ }
+ }
+ Ok(Self { filters, masks })
}
pub(crate) fn is_empty(&self) -> bool {
- self.filters.is_empty()
+ self.filters.is_empty() && self.masks.is_empty()
+ }
+
+ /// Table-schema indices of masked columns.
+ pub(crate) fn masked_columns(&self) -> HashSet<usize> {
+ self.masks.iter().map(|m| m.column).collect()
}
/// Table-schema indices the row filter reads.
@@ -79,6 +139,66 @@ impl Rules {
}
}
+/// Table-schema indices a mask reads.
+pub(crate) fn mask_inputs(mask: &ColumnMask) -> HashSet<usize> {
+ let mut out = HashSet::new();
+ mask.transform.collect_field_indices(&mut out);
+ out
+}
+
+/// Preserve the target's type and nullability.
+fn check_mask_fits(transform: &Transform, target: &DataField, fields:
&[DataField]) -> Result<()> {
+ let column_type = crate::arrow::paimon_type_to_arrow(target.data_type())?;
+ if let Some(output) = mask_output_type(transform, fields) {
+ if output != column_type {
+ return Err(unsupported(&format!(
+ "the mask on '{}' produces {output:?}, but the column is
{column_type:?}",
+ target.name()
+ )));
+ }
+ }
+ if !target.data_type().is_nullable() && mask_can_be_null(transform,
fields) {
+ return Err(unsupported(&format!(
+ "the mask on '{}' can produce null, but the column is NOT NULL",
+ target.name()
+ )));
+ }
+ if let Transform::Cast(index, to) = transform {
+ let from = fields[*index].data_type();
+ if !cast_agrees_with_java(from, to) {
+ return Err(unsupported(&format!(
+ "the mask on '{}' casts {from:?} to {to:?}, which this client
does not convert as \
+ Java does",
+ target.name()
+ )));
+ }
+ }
+ Ok(())
+}
+
+/// Arrow and Java disagree on casts such as INT -> TIMESTAMP.
+fn cast_agrees_with_java(from: &crate::spec::DataType, to:
&crate::spec::DataType) -> bool {
+ use crate::spec::DataType as Paimon;
+ let string = |t: &Paimon| matches!(t, Paimon::Char(_) |
Paimon::VarChar(_));
+ let bytes = |t: &Paimon| matches!(t, Paimon::Binary(_) |
Paimon::VarBinary(_));
+ let integer_bits = |t: &Paimon| match t {
+ Paimon::TinyInt(_) => Some(8),
+ Paimon::SmallInt(_) => Some(16),
+ Paimon::Int(_) => Some(32),
+ Paimon::BigInt(_) => Some(64),
+ _ => None,
+ };
+ let same = matches!(
+ (from.copy_with_nullable(true), to.copy_with_nullable(true)),
+ (Ok(from), Ok(to)) if from == to
+ );
+ let widening = matches!((integer_bits(from), integer_bits(to)), (Some(a),
Some(b)) if a < b);
+ same || widening
+ || string(to)
+ && (string(from) || integer_bits(from).is_some() || matches!(from,
Paimon::Boolean(_)))
+ || bytes(to) && (bytes(from) || string(from))
+}
+
fn mentioned_system_column(json: &str) -> Option<&'static str> {
[
crate::spec::ROW_ID_FIELD_NAME,
@@ -229,6 +349,288 @@ fn eval_err(e: &dyn std::fmt::Display) -> crate::Error {
unsupported(&format!("cannot evaluate the server's rules: {e}"))
}
+// Column masking
+
+/// Replace every copy of a masked column using the original batch's values.
+pub(crate) fn mask_batch(
+ batch: &RecordBatch,
+ masks: &[ColumnMask],
+ schema_fields: &[DataField],
+ batch_fields: &[DataField],
+) -> Result<RecordBatch> {
+ if masks.is_empty() {
+ return Ok(batch.clone());
+ }
+ let input = |index: usize| -> Result<ArrayRef> {
+ let field = schema_fields
+ .get(index)
+ .ok_or_else(|| unsupported("the server's mask reads an unknown
column"))?;
+ column_of(batch, field, batch_fields)
+ };
+ let mut columns = batch.columns().to_vec();
+ for mask in masks {
+ let field = &schema_fields[mask.column];
+ // By id: the caller may project the column twice.
+ let targets = batch_fields
+ .iter()
+ .enumerate()
+ .filter_map(|(position, f)| (f.id() ==
field.id()).then_some(position));
+ let Some(first) = targets.clone().next() else {
+ return Err(unsupported(&format!(
+ "the read does not carry '{}', which the server masks",
+ field.name()
+ )));
+ };
+ let target_type = batch.schema().field(first).data_type().clone();
+ let masked: ArrayRef = match &mask.transform {
+ Transform::Null => arrow_array::new_null_array(&target_type,
batch.num_rows()),
+ Transform::FieldRef(index) => input(*index)?,
+ Transform::Cast(index, to) => {
+ let cast = cast_to(&input(*index)?,
&crate::arrow::paimon_type_to_arrow(to)?)?;
+ // Arrow strings do not enforce CHAR/VARCHAR width.
+ apply_declared_width(&cast, to)
+ }
+ Transform::Upper(_)
+ | Transform::Lower(_)
+ | Transform::Concat(_)
+ | Transform::ConcatWs(_) => string_mask(batch, &mask.transform,
&input)?,
+ };
+ let masked = cast_to(&masked, &target_type)?;
+ for position in targets {
+ columns[position] = Arc::clone(&masked);
+ }
+ }
+ RecordBatch::try_new_with_options(
+ batch.schema(),
+ columns,
+
&arrow_array::RecordBatchOptions::new().with_row_count(Some(batch.num_rows())),
+ )
+ .map_err(|e| eval_err(&e))
+}
+
+fn mask_can_be_null(transform: &Transform, fields: &[DataField]) -> bool {
+ let nullable = |input: &TransformInput| match input {
+ TransformInput::Literal(literal) => literal.is_none(),
+ TransformInput::Field(index) =>
fields[*index].data_type().is_nullable(),
+ };
+ match transform {
+ Transform::Null => true,
+ Transform::FieldRef(index) | Transform::Cast(index, _) => {
+ fields[*index].data_type().is_nullable()
+ }
+ Transform::Upper(inputs) | Transform::Lower(inputs) |
Transform::Concat(inputs) => {
+ inputs.iter().any(nullable)
+ }
+ // Only a null separator makes CONCAT_WS null.
+ Transform::ConcatWs(inputs) => inputs.first().is_some_and(nullable),
+ }
+}
+
+/// `None` lets a NULL mask adopt the target's type.
+fn mask_output_type(transform: &Transform, fields: &[DataField]) ->
Option<arrow_schema::DataType> {
+ match transform {
+ Transform::Null => None,
+ Transform::FieldRef(index) => {
+ crate::arrow::paimon_type_to_arrow(fields[*index].data_type()).ok()
+ }
+ Transform::Cast(_, to) => crate::arrow::paimon_type_to_arrow(to).ok(),
+ Transform::Upper(_)
+ | Transform::Lower(_)
+ | Transform::Concat(_)
+ | Transform::ConcatWs(_) => Some(arrow_schema::DataType::Utf8),
+ }
+}
+
+/// Match Java's truncation and fixed-width padding.
+fn apply_declared_width(array: &ArrayRef, to: &crate::spec::DataType) ->
ArrayRef {
+ use crate::spec::DataType as Paimon;
+ match to {
+ Paimon::VarChar(t) => fit_strings(array, t.length() as usize, false),
+ Paimon::Char(t) => fit_strings(array, t.length(), true),
+ Paimon::VarBinary(t) => fit_bytes(array, t.length() as usize, false),
+ Paimon::Binary(t) => fit_bytes(array, t.length(), true),
+ _ => Arc::clone(array),
+ }
+}
+
+/// Counts characters, as Java does.
+fn fit_strings(array: &ArrayRef, length: usize, pad: bool) -> ArrayRef {
+ let Some(strings) =
array.as_any().downcast_ref::<arrow_array::StringArray>() else {
+ return Arc::clone(array);
+ };
+ let fitted: arrow_array::StringArray = strings
+ .iter()
+ .map(|value| {
+ value.map(|value| {
+ let chars = value.chars().count();
+ if chars > length {
+ value.chars().take(length).collect::<String>()
+ } else if pad && chars < length {
+ format!("{value}{}", " ".repeat(length - chars))
+ } else {
+ value.to_string()
+ }
+ })
+ })
+ .collect();
+ Arc::new(fitted)
+}
+
+fn fit_bytes(array: &ArrayRef, length: usize, pad: bool) -> ArrayRef {
+ let Some(bytes) =
array.as_any().downcast_ref::<arrow_array::BinaryArray>() else {
+ return Arc::clone(array);
+ };
+ let fitted: arrow_array::BinaryArray = bytes
+ .iter()
+ .map(|value| {
+ value.map(|value| {
+ let mut value = value.to_vec();
+ if value.len() > length || pad {
+ value.resize(length, 0);
+ }
+ value
+ })
+ })
+ .collect();
+ Arc::new(fitted)
+}
+
+/// Fail on invalid casts, as Java does.
+fn cast_to(array: &ArrayRef, to: &arrow_schema::DataType) -> Result<ArrayRef> {
+ if array.data_type() == to {
+ return Ok(Arc::clone(array));
+ }
+ let options = arrow_cast::CastOptions {
+ safe: false,
+ ..Default::default()
+ };
+ arrow_cast::cast_with_options(array, to, &options).map_err(|e|
eval_err(&e))
+}
+
+enum StringInput<'a> {
+ Literal(Option<&'a str>),
+ Column(arrow_array::StringArray),
+}
+
+impl StringInput<'_> {
+ fn value(&self, row: usize) -> Option<&str> {
+ match self {
+ Self::Literal(value) => *value,
+ Self::Column(values) => values.is_valid(row).then(||
values.value(row)),
+ }
+ }
+}
+
+fn string_mask(
+ batch: &RecordBatch,
+ transform: &Transform,
+ input: &dyn Fn(usize) -> Result<ArrayRef>,
+) -> Result<ArrayRef> {
+ use arrow_array::builder::StringBuilder;
+ use std::fmt::Write;
+
+ let inputs = match transform {
+ Transform::Upper(inputs) | Transform::Lower(inputs) if inputs.len() ==
1 => inputs,
+ Transform::Concat(inputs) => inputs,
+ Transform::ConcatWs(inputs) if inputs.len() >= 2 => inputs,
+ _ => return Err(unsupported("the server's string mask has the wrong
inputs")),
+ };
+ let resolved: Vec<StringInput<'_>> = inputs
+ .iter()
+ .map(|i| match i {
+ TransformInput::Literal(literal) =>
Ok(StringInput::Literal(literal.as_deref())),
+ TransformInput::Field(index) => Ok(StringInput::Column(
+ cast_to(&input(*index)?, &arrow_schema::DataType::Utf8)?
+ .as_string::<i32>()
+ .clone(),
+ )),
+ })
+ .collect::<Result<_>>()?;
+ if matches!(transform, Transform::Upper(_) | Transform::Lower(_)) {
+ let upper = matches!(transform, Transform::Upper(_));
+ let literal = match &resolved[0] {
+ StringInput::Column(array) => return
Ok(Arc::new(string_case(array, upper))),
+ StringInput::Literal(value) => value,
+ };
+ let mut values = StringBuilder::with_capacity(batch.num_rows(), 0);
+ match literal {
+ Some(value) => values.append_value_n(
+ if upper {
+ value.to_uppercase()
+ } else {
+ value.to_lowercase()
+ },
+ batch.num_rows(),
+ ),
+ None => values.append_nulls(batch.num_rows()),
+ }
+ return Ok(Arc::new(values.finish()));
+ }
+ let mut values = StringBuilder::with_capacity(batch.num_rows(), 0);
+ for row in 0..batch.num_rows() {
+ let (parts, separator) = if matches!(transform,
Transform::ConcatWs(_)) {
+ let Some(separator) = resolved[0].value(row) else {
+ values.append_null();
+ continue;
+ };
+ (&resolved[1..], separator)
+ } else {
+ if resolved.iter().any(|value| value.value(row).is_none()) {
+ values.append_null();
+ continue;
+ }
+ (resolved.as_slice(), "")
+ };
+ let mut first = true;
+ for part in parts.iter().filter_map(|value| value.value(row)) {
+ if !first {
+ values
+ .write_str(separator)
+ .expect("infallible string write");
+ }
+ values.write_str(part).expect("infallible string write");
+ first = false;
+ }
+ values.append_value("");
+ }
+ Ok(Arc::new(values.finish()))
+}
+
+fn string_case(array: &arrow_array::StringArray, upper: bool) ->
arrow_array::StringArray {
+ let offsets = array.value_offsets();
+ let start = offsets[0];
+ let bytes = &array.value_data()[start as usize..offsets[array.len()] as
usize];
+ if bytes.is_ascii() {
+ let mut values = bytes.to_vec();
+ if upper {
+ values.make_ascii_uppercase();
+ } else {
+ values.make_ascii_lowercase();
+ }
+ let offsets = if start == 0 {
+ array.offsets().clone()
+ } else {
+ arrow_buffer::OffsetBuffer::new(
+ offsets.iter().map(|o| o - start).collect::<Vec<_>>().into(),
+ )
+ };
+ return arrow_array::StringArray::new(offsets, values.into(),
array.nulls().cloned());
+ }
+ // Unicode lowercasing needs context, e.g. the final Greek sigma.
+ array
+ .iter()
+ .map(|value| {
+ value.map(|value| {
+ if upper {
+ value.to_uppercase()
+ } else {
+ value.to_lowercase()
+ }
+ })
+ })
+ .collect()
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -237,8 +639,8 @@ mod tests {
IntType, PredicateBuilder, PredicateOperator, TimestampType,
VarBinaryType, VarCharType,
};
use arrow_array::{
- Date32Array, Decimal128Array, Int32Array, Int64Array,
RecordBatchOptions, StringArray,
- TimestampMillisecondArray,
+ BinaryArray, Date32Array, Decimal128Array, Int16Array, Int32Array,
Int64Array, Int8Array,
+ RecordBatchOptions, StringArray, TimestampMillisecondArray,
};
use arrow_schema::{Field, Schema};
@@ -303,11 +705,36 @@ mod tests {
)
}
+ /// A Java `FieldRef`, as a transform input.
+ fn input(field: &str) -> String {
+ format!(r#"{{"index":0,"name":"{field}","type":"STRING"}}"#)
+ }
+
+ fn field_ref(field: &str) -> String {
+ format!(r#"{{"name":"FIELD_REF","fieldRef":{}}}"#, input(field))
+ }
+
+ fn cast(field: &str, to: &str) -> String {
+ format!(
+ r#"{{"name":"CAST","fieldRef":{},"type":"{to}"}}"#,
+ input(field)
+ )
+ }
+
+ fn string_transform(name: &str, inputs: &[&str]) -> String {
+ format!(r#"{{"name":"{name}","inputs":[{}]}}"#, inputs.join(","))
+ }
+
fn ids(batch: &RecordBatch) -> Vec<i32> {
let ids = batch.column(0).as_any().downcast_ref::<Int32Array>();
ids.unwrap().values().to_vec()
}
+ fn strings(batch: &RecordBatch, column: usize) -> Vec<Option<&str>> {
+ let strings =
batch.column(column).as_any().downcast_ref::<StringArray>();
+ strings.unwrap().iter().collect()
+ }
+
// ---------------------------------------------------------------------
// Parsing
// ---------------------------------------------------------------------
@@ -345,9 +772,26 @@ mod tests {
}
#[test]
- fn test_any_mask_is_refused() {
- let message = refusal(parse(&[], &[("name", NULL)], &rule_fields()));
- assert!(message.contains("column masking"), "{message}");
+ fn test_parse_reads_the_masks() {
+ let fields = rule_fields();
+ let upper = string_transform("UPPER", &[&input("name")]);
+ let rules = parse(&[], &[("name", &upper), ("id", NULL)],
&fields).unwrap();
+ // Masks follow schema order.
+ assert_eq!(
+ rules.masks,
+ vec![
+ ColumnMask {
+ column: 0,
+ transform: Transform::Null
+ },
+ ColumnMask {
+ column: 1,
+ transform: Transform::Upper(vec![TransformInput::Field(1)])
+ },
+ ]
+ );
+ assert_eq!(rules.masked_columns(), HashSet::from([0, 1]));
+ assert_eq!(mask_inputs(&rules.masks[1]), HashSet::from([1]));
}
#[test]
@@ -370,6 +814,49 @@ 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();
+ let missing_input = field_ref("missing");
+ for (column, mask, reason) in [
+ ("", NULL, "empty column mask"),
+ ("name", "", "empty column mask"),
+ ("name", "null", "cannot parse"),
+ ("name", "{", "cannot parse"),
+ ("name", r#"{"name":"ROT13"}"#, "cannot parse"),
+ ("missing", NULL, "'missing', which is not a column"),
+ ("name", &missing_input, "unknown field `missing`"),
+ ] {
+ let message = refusal(parse(&[], &[(column, mask)], &fields));
+ assert!(message.contains(reason), "{column:?} {mask:?}:
{message}");
+ }
+ }
+
#[test]
fn test_a_filter_on_a_system_column_is_refused() {
let fields = rule_fields();
@@ -382,6 +869,98 @@ mod tests {
}
}
+ #[test]
+ fn test_a_mask_on_a_system_column_is_inert() {
+ let fields = rule_fields();
+ for column in [
+ "_ROW_ID",
+ "_SEQUENCE_NUMBER",
+ "_VALUE_KIND",
+ "_LEVEL",
+ "rowkind",
+ ] {
+ let rules = parse(&[], &[(column, NULL)], &fields).unwrap();
+ assert!(rules.is_empty(), "{column}");
+ }
+ // `_KEY_` names are not exempt system fields.
+ let message = refusal(parse(&[], &[("_KEY_id", NULL)], &fields));
+ assert!(
+ message.contains("'_KEY_id', which is not a column"),
+ "{message}"
+ );
+ // Skipped masks must still parse.
+ for mask in ["null", "{"] {
+ let message = refusal(parse(&[], &[("_ROW_ID", mask)], &fields));
+ assert!(message.contains("cannot parse"), "{mask}: {message}");
+ }
+ }
+
+ #[test]
+ fn test_a_mask_must_keep_the_column_type() {
+ let fields = rule_fields();
+ for (column, mask) in [
+ ("id", string_transform("UPPER", &[&input("name")])),
+ ("id", string_transform("CONCAT", &[r#""x""#])),
+ ("id", field_ref("name")),
+ ("id", cast("id", "STRING")),
+ ("id", cast("id", "BIGINT")),
+ ("name", cast("name", "BINARY(3)")),
+ ] {
+ let message = refusal(parse(&[], &[(column, &mask)], &fields));
+ assert!(message.contains("produces"), "{mask}: {message}");
+ }
+ for (column, mask) in [
+ ("id", NULL.to_string()),
+ ("name", cast("id", "VARCHAR(2)")),
+ ("bin", cast("name", "BINARY(3)")),
+ ] {
+ assert!(parse(&[], &[(column, &mask)], &fields).is_ok(), "{mask}");
+ }
+ }
+
+ #[test]
+ fn test_a_not_null_column_takes_only_a_mask_that_cannot_be_null() {
+ let fields = rule_fields();
+ let (name, nn) = (input("name"), input("nn"));
+ for (mask, fits) in [
+ (NULL.to_string(), false),
+ (field_ref("name"), false),
+ (field_ref("nn"), true),
+ (cast("name", "STRING"), false),
+ (cast("nn", "VARCHAR(3)"), true),
+ (string_transform("UPPER", &[&nn]), true),
+ (string_transform("CONCAT", &[&nn, "null"]), false),
+ (string_transform("CONCAT_WS", &["null", &nn]), false),
+ // Only the separator can make CONCAT_WS null.
+ (string_transform("CONCAT_WS", &[r#""-""#, &nn, &name]), true),
+ ] {
+ let parsed = parse(&[], &[("nn", &mask)], &fields);
+ if fits {
+ assert!(parsed.is_ok(), "{mask}: {parsed:?}");
+ } else {
+ assert!(refusal(parsed).contains("can produce null"),
"{mask}");
+ }
+ }
+ }
+
+ #[test]
+ fn test_a_mask_may_not_read_another_masked_column() {
+ let fields = rule_fields();
+ let upper_name = string_transform("UPPER", &[&input("name")]);
+ assert!(parse(&[], &[("name", &upper_name)], &fields).is_ok());
+ assert!(parse(&[], &[("name", &upper_name), ("id", NULL)],
&fields).is_ok());
+ // `alias` would expose the unmasked `name`.
+ let message = refusal(parse(
+ &[],
+ &[("name", NULL), ("alias", &upper_name)],
+ &fields,
+ ));
+ assert!(
+ message.contains("reads 'name', which is masked too"),
+ "{message}"
+ );
+ }
+
// ---------------------------------------------------------------------
// Row filter
// ---------------------------------------------------------------------
@@ -696,4 +1275,346 @@ mod tests {
3
);
}
+
+ // Column masking
+
+ fn mask_rows() -> RecordBatch {
+ let bytes: Vec<Option<&[u8]>> = vec![Some(&[1, 2, 3, 4, 5, 6]),
Some(&[1, 2]), None];
+ batch(
+ &rule_fields(),
+ vec![
+ Arc::new(Int32Array::from(vec![Some(12345), Some(7), None])),
+ Arc::new(StringArray::from(vec![
+ Some("supersecret"),
+ Some("ab"),
+ None,
+ ])),
+ Arc::new(StringArray::from(vec![Some("MiXed"), None,
Some("x")])),
+ Arc::new(BinaryArray::from(bytes)),
+ Arc::new(StringArray::from(vec!["héllo wörld", "q", "z"])),
+ ],
+ )
+ }
+
+ fn masked(column: &str, mask: &str) -> RecordBatch {
+ let fields = rule_fields();
+ let rules = parse(&[], &[(column, mask)], &fields).unwrap();
+ mask_batch(&mask_rows(), &rules.masks, &fields, &fields).unwrap()
+ }
+
+ #[test]
+ fn test_string_masks_propagate_nulls_as_java_does() {
+ let (name, alias) = (input("name"), input("alias"));
+ let cases = [
+ (NULL.to_string(), vec![None, None, None]),
+ (field_ref("alias"), vec![Some("MiXed"), None, Some("x")]),
+ (
+ string_transform("UPPER", &[&alias]),
+ vec![Some("MIXED"), None, Some("X")],
+ ),
+ (
+ string_transform("LOWER", &[&alias]),
+ vec![Some("mixed"), None, Some("x")],
+ ),
+ (
+ string_transform("UPPER", &[r#""Straße""#]),
+ vec![Some("STRASSE"); 3],
+ ),
+ (string_transform("LOWER", &[r#""ΟΣ""#]), vec![Some("ος"); 3]),
+ (string_transform("UPPER", &["null"]), vec![None; 3]),
+ (string_transform("CONCAT", &[]), vec![Some(""); 3]),
+ (
+ string_transform("CONCAT", &[&name, r#""-""#, &alias]),
+ vec![Some("supersecret-MiXed"), None, None],
+ ),
+ (
+ string_transform("CONCAT", &[r#""****""#]),
+ vec![Some("****"); 3],
+ ),
+ (string_transform("CONCAT", &[&name, "null"]), vec![None; 3]),
+ // CONCAT_WS skips null payloads and is null on a null separator.
+ (
+ string_transform("CONCAT_WS", &[r#""-""#, r#""x""#, &name,
"null", &alias]),
+ vec![Some("x-supersecret-MiXed"), Some("x-ab"), Some("x-x")],
+ ),
+ (
+ string_transform("CONCAT_WS", &[&alias, &name, r#""y""#]),
+ vec![Some("supersecretMiXedy"), None, Some("y")],
+ ),
+ (
+ string_transform("CONCAT_WS", &[r#""-""#, "null", "null"]),
+ vec![Some(""); 3],
+ ),
+ ];
+ for (mask, expected) in cases {
+ let out = masked("name", &mask);
+ assert_eq!(strings(&out, 1), expected, "{mask}");
+ assert_eq!(strings(&out, 2), [Some("MiXed"), None, Some("x")],
"{mask}");
+ }
+ }
+
+ #[test]
+ fn test_a_cast_java_converts_differently_is_refused() {
+ let fields = vec![
+ field(0, "n", DataType::BigInt(crate::spec::BigIntType::new())),
+ field(
+ 1,
+ "ts",
+
DataType::Timestamp(crate::spec::TimestampType::new(3).unwrap()),
+ ),
+ field(2, "d", DataType::Double(crate::spec::DoubleType::new())),
+ field(3, "s", string()),
+ field(4, "b",
DataType::VarBinary(VarBinaryType::new(10).unwrap())),
+ field(5, "i", DataType::Int(IntType::new())),
+ ];
+ for (target, mask) in [
+ ("ts", cast("n", "TIMESTAMP(3)")),
+ ("s", cast("d", "VARCHAR(10)")),
+ ("s", cast("b", "VARCHAR(10)")),
+ // Java trims the string first.
+ ("n", cast("s", "BIGINT")),
+ ("i", cast("n", "INT")),
+ ] {
+ let message = refusal(parse(&[], &[(target, &mask)], &fields));
+ assert!(message.contains("as Java does"), "{mask}: {message}");
+ }
+ for (target, mask) in [
+ ("s", cast("n", "VARCHAR(10)")),
+ ("b", cast("s", "VARBINARY(10)")),
+ ("ts", cast("ts", "TIMESTAMP(3)")),
+ ] {
+ assert!(parse(&[], &[(target, &mask)], &fields).is_ok(), "{mask}");
+ }
+ }
+
+ #[test]
+ fn test_integer_cast_masks_widen_exactly() {
+ let tiny: ArrayRef = Arc::new(Int8Array::from(vec![Some(i8::MIN),
None, Some(i8::MAX)]));
+ let small: ArrayRef =
+ Arc::new(Int16Array::from(vec![Some(i16::MIN), None,
Some(i16::MAX)]));
+ let int: ArrayRef = Arc::new(Int32Array::from(vec![Some(i32::MIN),
None, Some(i32::MAX)]));
+ let cases: [(&str, &str, ArrayRef, ArrayRef); 6] = [
+ (
+ "TINYINT",
+ "SMALLINT",
+ tiny.clone(),
+ Arc::new(Int16Array::from(vec![Some(-128), None, Some(127)])),
+ ),
+ (
+ "TINYINT",
+ "INT",
+ tiny.clone(),
+ Arc::new(Int32Array::from(vec![Some(-128), None, Some(127)])),
+ ),
+ (
+ "TINYINT",
+ "BIGINT",
+ tiny,
+ Arc::new(Int64Array::from(vec![Some(-128), None, Some(127)])),
+ ),
+ (
+ "SMALLINT",
+ "INT",
+ small.clone(),
+ Arc::new(Int32Array::from(vec![Some(-32768), None,
Some(32767)])),
+ ),
+ (
+ "SMALLINT",
+ "BIGINT",
+ small,
+ Arc::new(Int64Array::from(vec![Some(-32768), None,
Some(32767)])),
+ ),
+ (
+ "INT",
+ "BIGINT",
+ int,
+ Arc::new(Int64Array::from(vec![
+ Some(-2147483648),
+ None,
+ Some(2147483647),
+ ])),
+ ),
+ ];
+ for (from, to, values, expected) in cases {
+ let fields = vec![
+ field(
+ 0,
+ "input",
+ serde_json::from_value(serde_json::json!(from)).unwrap(),
+ ),
+ field(
+ 1,
+ "wide",
+ serde_json::from_value(serde_json::json!(to)).unwrap(),
+ ),
+ ];
+ let target =
+ arrow_cast::cast(&Int64Array::from(vec![999; 3]),
expected.data_type()).unwrap();
+ let rows = batch(&fields, vec![values, target]);
+ let mask = serde_json::json!({"name":"CAST", "fieldRef": {
+ "index":0, "name":"input", "type":from,
+ }, "type":to})
+ .to_string();
+ let rules = parse(&[], &[("wide", &mask)], &fields).unwrap();
+ let out = mask_batch(&rows, &rules.masks, &fields,
&fields).unwrap();
+ assert_eq!(out.column(1), &expected, "{from} -> {to}");
+ assert_eq!(out.column(0), rows.column(0));
+ }
+ }
+
+ #[test]
+ fn test_string_masks_preserve_unicode_and_sliced_inputs() {
+ let fields = vec![field(0, "value", string())];
+ let input = serde_json::json!({"index":0, "name":"value",
"type":"STRING"}).to_string();
+ for (values, upper, lower) in [
+ (
+ vec![
+ Some("prefix"),
+ Some("aZ"),
+ None,
+ Some(""),
+ Some("q"),
+ Some("suffix"),
+ ],
+ vec![Some("AZ"), None, Some(""), Some("Q")],
+ vec![Some("az"), None, Some(""), Some("q")],
+ ),
+ (
+ vec![
+ Some("prefix"),
+ Some("ΟΣ"),
+ Some("Straße"),
+ None,
+ Some("İ"),
+ Some("suffix"),
+ ],
+ vec![Some("ΟΣ"), Some("STRASSE"), None, Some("İ")],
+ vec![Some("ος"), Some("straße"), None, Some("i\u{307}")],
+ ),
+ (vec![None; 6], vec![None; 4], vec![None; 4]),
+ ] {
+ let rows = batch(
+ &fields,
+ vec![Arc::new(StringArray::from(values).slice(1, 4))],
+ );
+ for (operation, expected) in [("UPPER", upper), ("LOWER", lower)] {
+ let mask = string_transform(operation, &[&input]);
+ let rules = parse(&[], &[("value", &mask)], &fields).unwrap();
+ let out = mask_batch(&rows, &rules.masks, &fields,
&fields).unwrap();
+ assert_eq!(strings(&out, 0), expected);
+ let empty = mask_batch(&rows.slice(2, 0), &rules.masks,
&fields, &fields).unwrap();
+ assert_eq!(empty.num_rows(), 0);
+ }
+ }
+ }
+
+ #[test]
+ fn test_a_cast_mask_applies_the_declared_width() {
+ for (mask, expected) in [
+ (
+ cast("name", "VARCHAR(3)"),
+ vec![Some("sup"), Some("ab"), None],
+ ),
+ (
+ cast("name", "CHAR(5)"),
+ vec![Some("super"), Some("ab "), None],
+ ),
+ // Characters, not bytes.
+ (
+ cast("nn", "VARCHAR(3)"),
+ vec![Some("hél"), Some("q"), Some("z")],
+ ),
+ (cast("id", "VARCHAR(2)"), vec![Some("12"), Some("7"), None]),
+ ] {
+ assert_eq!(strings(&masked("name", &mask), 1), expected, "{mask}");
+ }
+ let cases: [(String, Vec<Option<&[u8]>>); 3] = [
+ (
+ cast("bin", "BINARY(4)"),
+ vec![Some(&[1, 2, 3, 4]), Some(&[1, 2, 0, 0]), None],
+ ),
+ (
+ cast("bin", "VARBINARY(4)"),
+ vec![Some(&[1, 2, 3, 4]), Some(&[1, 2]), None],
+ ),
+ (
+ cast("name", "BINARY(3)"),
+ vec![Some(b"sup"), Some(b"ab\0"), None],
+ ),
+ ];
+ for (mask, expected) in cases {
+ let out = masked("bin", &mask);
+ let bytes = out.column(3).as_any().downcast_ref::<BinaryArray>();
+ assert_eq!(
+ bytes.unwrap().iter().collect::<Vec<_>>(),
+ expected,
+ "{mask}"
+ );
+ }
+ }
+
+ #[test]
+ fn test_every_copy_of_a_masked_column_is_masked() {
+ let fields = rule_fields();
+ let rows = mask_rows().project(&[1, 0, 1]).unwrap();
+ let rows_fields = [fields[1].clone(), fields[0].clone(),
fields[1].clone()];
+ let mask = string_transform("CONCAT", &[&input("name"), r#""!""#]);
+ let rules = parse(&[], &[("name", &mask)], &fields).unwrap();
+ let out = mask_batch(&rows, &rules.masks, &fields,
&rows_fields).unwrap();
+ for column in [0, 2] {
+ assert_eq!(
+ strings(&out, column),
+ [Some("supersecret!"), Some("ab!"), None]
+ );
+ }
+ assert_eq!(out.column(1), rows.column(1));
+ }
+
+ #[test]
+ fn test_masks_read_the_batch_as_read() {
+ // Bypass parsing to verify both masks read the original batch.
+ let fields = rule_fields();
+ let masks = [
+ ColumnMask {
+ column: 0,
+ transform: Transform::Null,
+ },
+ ColumnMask {
+ column: 1,
+ transform: Transform::Cast(0, string()),
+ },
+ ];
+ let out = mask_batch(&mask_rows(), &masks, &fields, &fields).unwrap();
+ assert_eq!(out.column(0).null_count(), 3);
+ assert_eq!(strings(&out, 1), [Some("12345"), Some("7"), None]);
+ }
+
+ #[test]
+ fn test_a_mask_refuses_a_batch_without_its_columns() {
+ let fields = rule_fields();
+ let rows = mask_rows();
+ let mask = field_ref("alias");
+ let masks = parse(&[], &[("name", &mask)], &fields).unwrap().masks;
+ let without_target = rows.project(&[0, 2]).unwrap();
+ let message = refusal(mask_batch(
+ &without_target,
+ &masks,
+ &fields,
+ &[fields[0].clone(), fields[2].clone()],
+ ));
+ assert!(message.contains("does not carry 'name'"), "{message}");
+ let without_input = rows.project(&[0, 1]).unwrap();
+ let message = refusal(mask_batch(&without_input, &masks, &fields,
&fields[..2]));
+ assert!(message.contains("does not carry 'alias'"), "{message}");
+ }
+
+ #[test]
+ fn test_no_mask_keeps_a_zero_column_batch_whole() {
+ let options = RecordBatchOptions::new().with_row_count(Some(3));
+ let empty =
+ RecordBatch::try_new_with_options(Arc::new(Schema::empty()),
Vec::new(), &options)
+ .unwrap();
+ let out = mask_batch(&empty, &[], &rule_fields(), &[]).unwrap();
+ assert_eq!(out.num_rows(), 3);
+ }
}
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index ea327b31..42ee2f3b 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -46,7 +46,7 @@ struct NormalizedFilter {
/// Exact filters are fully enforced by paimon-core scan planning using only
/// partition-owned semantics, without requiring residual filtering above the
/// scan.
-fn is_exact_filter_pushdown_for_schema(
+pub(super) fn is_exact_filter_pushdown_for_schema(
fields: &[DataField],
partition_keys: &[String],
filter: &Predicate,
@@ -65,7 +65,8 @@ pub(super) fn split_scan_predicates(
filter: Predicate,
) -> (Option<Predicate>, Vec<Predicate>) {
let partition_keys = table.schema().partition_keys();
- if partition_keys.is_empty() {
+ // Auth rules may mask partition keys; retain the full read predicate.
+ if partition_keys.is_empty() ||
table.schema().core_options().query_auth_enabled() {
(None, filter.split_and())
} else {
split_partition_and_data_predicates(filter, table.schema().fields(),
partition_keys)
@@ -448,11 +449,13 @@ impl<'a> PaimonReadBuilder<'a> {
/// Exact filters are fully enforced by paimon-core scan planning, without
/// requiring residual filtering above the scan.
pub fn is_exact_filter_pushdown(&self, filter: &Predicate) -> bool {
- is_exact_filter_pushdown_for_schema(
- self.table.schema().fields(),
- self.table.schema().partition_keys(),
- filter,
- )
+ // Masked partition predicates may need a read-side residual.
+ !self.table.schema().core_options().query_auth_enabled()
+ && is_exact_filter_pushdown_for_schema(
+ self.table.schema().fields(),
+ self.table.schema().partition_keys(),
+ filter,
+ )
}
/// Set row ID ranges `[from, to]` (inclusive). An empty vector selects no
rows.
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index d68ff2a3..11d6bc3f 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -1679,7 +1679,10 @@ mod tests {
Some(Arc::new(crate::table::query_auth::QueryAuthGrant::new(
table.query_auth_session().unwrap(),
None,
- crate::table::query_auth::Rules { filters },
+ crate::table::query_auth::Rules {
+ filters,
+ masks: Vec::new(),
+ },
)))
};
let s = || split(vec![file("a", 10, None)], true);
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 183984bd..94c1ae8b 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -1069,20 +1069,35 @@ impl<'a> PaimonTableRead<'a> {
}
}
- /// Reads what the row filter needs, filters on stored values and projects
- /// back (Java `doAuth`).
+ /// Filter rows, apply masks, then evaluate predicates on masked columns.
fn read_restricted(
&self,
data_splits: &[DataSplit],
grant: &super::query_auth::QueryAuthGrant,
) -> crate::Result<ArrowRecordBatchStream> {
- use super::query_auth::{filter_batch, unsupported};
+ use super::query_auth::{filter_batch, mask_batch, mask_inputs,
unsupported};
+ use crate::arrow::format::FilePredicates;
+ use crate::arrow::residual::{filter_record_batch_by_predicates,
widen_scan_fields};
+ use std::collections::HashSet;
let schema_fields = self.table.schema().fields().to_vec();
let rules = grant.rules();
super::query_auth::reject_throwing_extractions(&self.read_type)?;
+
+ let masked = rules.masked_columns();
+ let columns_of = |predicate: &Predicate| {
+ let mut columns = HashSet::new();
+ predicate.collect_leaf_field_indices(&mut columns);
+ columns
+ };
+ let (after_mask, raw): (Vec<Predicate>, Vec<Predicate>) = self
+ .data_predicates
+ .iter()
+ .cloned()
+ .partition(|p| !columns_of(p).is_disjoint(&masked));
+ let after_mask_columns: HashSet<usize> =
after_mask.iter().flat_map(columns_of).collect();
let index_of = |field: &DataField| schema_fields.iter().position(|s|
s.id() == field.id());
- let needed = rules.filter_columns();
+ let mut needed = rules.filter_columns();
let core_options = self.table.schema.core_options();
let view_fields = core_options.blob_view_fields();
if core_options.blob_view_resolve_enabled() &&
self.table.rest_env().is_some() {
@@ -1098,14 +1113,28 @@ impl<'a> PaimonTableRead<'a> {
)));
}
}
- // A partly projected column would feed the filter a partial value
(Java
- // `validateReadType`).
+ let visible: HashSet<usize> = self
+ .read_type
+ .iter()
+ .filter_map(index_of)
+ .chain(after_mask_columns.iter().copied())
+ .collect();
+ let masks: Vec<_> = rules
+ .masks
+ .iter()
+ .filter(|m| visible.contains(&m.column))
+ .cloned()
+ .collect();
+ needed.extend(&after_mask_columns);
+ masks.iter().for_each(|m| needed.extend(mask_inputs(m)));
+ // Rules require complete input columns.
for field in &self.read_type {
if let Some(index) = index_of(field) {
- if needed.contains(&index) && field.data_type() !=
schema_fields[index].data_type()
+ if (needed.contains(&index) || masked.contains(&index))
+ && field.data_type() != schema_fields[index].data_type()
{
return Err(unsupported(&format!(
- "the server's row filter reads '{}', which the read
projects only in part",
+ "the server's rules involve '{}', which the read
projects only in part",
field.name()
)));
}
@@ -1121,16 +1150,25 @@ impl<'a> PaimonTableRead<'a> {
physical.push(field.clone());
}
}
- let predicates = crate::arrow::format::FilePredicates {
+ let mut predicates = FilePredicates {
predicates: self.data_predicates.clone(),
row_filter_factory: None,
file_fields: schema_fields.clone(),
};
- physical = crate::arrow::residual::widen_scan_fields(&physical,
Some(&predicates));
+ physical = widen_scan_fields(&physical, Some(&predicates));
let mut inner = self.clone();
inner.read_type = physical.clone();
+ inner.data_predicates = raw;
inner.limit = None;
let has_blobs = physical.iter().any(|f|
f.data_type().is_blob_file_field());
+ let (before_mask, pending): (Vec<_>, Vec<_>) =
+ predicates.predicates.into_iter().partition(|predicate| {
+ has_blobs
+ && columns_of(predicate).iter().all(|i| {
+ !masked.contains(i) &&
!schema_fields[*i].data_type().is_blob_file_field()
+ })
+ });
+ predicates.predicates = before_mask;
let unresolved_table;
if has_blobs {
// Keep payload predicates above authorization too: they can open
@@ -1148,6 +1186,11 @@ impl<'a> PaimonTableRead<'a> {
let filters = rules.filters.clone();
let batch_fields = physical.clone();
+ let after_mask = FilePredicates {
+ predicates: after_mask,
+ row_filter_factory: None,
+ file_fields: schema_fields.clone(),
+ };
let stream = stream.map(move |batch| {
let batch = batch?;
let names_match = batch.num_columns() == physical.len()
@@ -1162,11 +1205,19 @@ impl<'a> PaimonTableRead<'a> {
"the read returned columns the rules cannot address",
));
}
- filter_batch(&batch, &filters, &schema_fields, &physical)
+ 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)?;
+ let batch = mask_batch(&batch, &masks, &schema_fields, &physical)?;
+ if has_blobs {
+ Ok(batch)
+ } else {
+ filter_record_batch_by_predicates(batch, &after_mask,
&physical)
+ }
});
let stream: ArrowRecordBatchStream = Box::pin(stream);
if has_blobs {
- return self.finish_authorized_blobs(stream, &batch_fields,
&core_options);
+ return self.finish_authorized_blobs(stream, &batch_fields,
&pending, &core_options);
}
let stream = project_authorized_stream(stream, self.read_type.len());
Ok(match self.limit {
@@ -1179,6 +1230,7 @@ impl<'a> PaimonTableRead<'a> {
&self,
stream: ArrowRecordBatchStream,
batch_fields: &[DataField],
+ predicates: &[Predicate],
core_options: &CoreOptions<'_>,
) -> crate::Result<ArrowRecordBatchStream> {
use super::managed_blob_reader::{resolve_primary_key_blob_stream,
ManagedBlobReadPlan};
@@ -1188,7 +1240,7 @@ impl<'a> PaimonTableRead<'a> {
&& self.table.rest_env().is_some()
&& batch_fields.iter().any(|f| view_fields.contains(f.name()));
let predicate_views: std::collections::HashSet<_> =
- super::query_auth::leaf_names(&self.data_predicates)
+ super::query_auth::leaf_names(predicates)
.intersection(&view_fields)
.cloned()
.collect();
@@ -1212,7 +1264,7 @@ impl<'a> PaimonTableRead<'a> {
// The ordinary Blob plan must not decode those payloads a second time.
let blob_plan = ManagedBlobReadPlan::new(
&self.read_type,
- &self.data_predicates,
+ predicates,
self.table.schema().fields(),
core_options,
);
@@ -1241,7 +1293,7 @@ impl<'a> PaimonTableRead<'a> {
}
None => {
let predicates = crate::arrow::format::FilePredicates {
- predicates: self.data_predicates.clone(),
+ predicates: predicates.to_vec(),
row_filter_factory: None,
file_fields: self.table.schema().fields().to_vec(),
};
@@ -2741,6 +2793,7 @@ mod tests {
} else {
Vec::new()
},
+ masks: Vec::new(),
},
)
}
@@ -2998,7 +3051,10 @@ mod tests {
let grant = crate::table::query_auth::QueryAuthGrant::new(
table.query_auth_session().unwrap(),
Some(Vec::new()),
- crate::table::query_auth::Rules { filters },
+ crate::table::query_auth::Rules {
+ filters,
+ masks: Vec::new(),
+ },
);
let split = split_with_grant(Some(grant));
let narrow = TableRead::new(&table, Vec::new(), Vec::new());
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index b56bf4a4..dd2fc4bb 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -1272,6 +1272,7 @@ impl<'a> TableScan<'a> {
/// `seed` must render as a decimal integer. Accepting any `ToString` seed
/// keeps ordinary Rust integer calls ergonomic while allowing language
/// bindings to preserve arbitrary-precision integer seeds.
+ /// Only predicates on unmasked partition keys are supported.
pub fn with_chunk_shuffle(self, seed: impl ToString, chunk_size: u64) ->
crate::Result<Self> {
let config = ChunkShuffle::from_decimal_seed(&seed.to_string(),
chunk_size)?;
match self.0 {
@@ -1293,7 +1294,13 @@ impl<'a> TableScan<'a> {
.to_string(),
});
}
- if !scan.data_predicates.is_empty() {
+ if !scan.data_predicates.is_empty()
+ &&
!super::read_builder::is_exact_filter_pushdown_for_schema(
+ scan.table.schema().fields(),
+ scan.table.schema().partition_keys(),
+ &Predicate::and(scan.data_predicates.clone()),
+ )
+ {
return Err(crate::Error::Unsupported {
message: "chunk_shuffle only supports partition
predicates".to_string(),
});
@@ -1720,10 +1727,7 @@ impl<'a> PaimonTableScan<'a> {
/// `scan.snapshot-id` / `scan.tag-name` handling.
pub async fn plan(&self) -> crate::Result<Plan> {
let grant = self.authorize_query().await?;
- let restricted = grant
- .as_deref()
- .filter(|grant| !grant.is_unrestricted())
- .map(|grant| self.restricted_by(grant));
+ let restricted = grant.as_deref().map(|grant|
self.restricted_by(grant));
let scan = restricted.as_ref().unwrap_or(self);
if let Some((start, end)) = scan
.table
@@ -1750,10 +1754,7 @@ impl<'a> PaimonTableScan<'a> {
/// Plan the full scan and return metadata-pruning trace counters.
pub async fn plan_with_trace(&self) -> crate::Result<(Plan, ScanTrace)> {
let grant = self.authorize_query().await?;
- let restricted = grant
- .as_deref()
- .filter(|grant| !grant.is_unrestricted())
- .map(|grant| self.restricted_by(grant));
+ let restricted = grant.as_deref().map(|grant|
self.restricted_by(grant));
let scan = restricted.as_ref().unwrap_or(self);
let mut trace = ScanTrace {
limit: scan.limit,
@@ -1793,20 +1794,54 @@ impl<'a> PaimonTableScan<'a> {
Ok((plan.planned(grant), trace))
}
- /// The scan a restricted grant plans: a limit cannot count rows the row
- /// filter drops.
+ /// Keep pruning off masked columns and apply limits after row filtering.
fn restricted_by(&self, grant: &super::query_auth::QueryAuthGrant) -> Self
{
let rules = grant.rules();
+ let fields = self.table.schema().fields();
+ let masked: HashSet<&str> = rules
+ .masks
+ .iter()
+ .map(|m| fields[m.column].name())
+ .collect();
+ // By name: the bucket predicate is indexed by bucket-key position.
+ let touches_masked = |predicate: &Predicate| {
+ super::query_auth::leaf_names(std::slice::from_ref(predicate))
+ .iter()
+ .any(|name| masked.contains(name.as_str()))
+ };
let mut scan = self.clone();
- if !rules.filters.is_empty() {
+ scan.data_predicates.retain(|p| !touches_masked(p));
+ if scan.bucket_predicate.as_ref().is_some_and(touches_masked) {
+ scan.bucket_predicate = None;
+ }
+ let dropped = scan.data_predicates.len() != self.data_predicates.len();
+ if !rules.filters.is_empty() || dropped {
scan.limit = scan.limit.filter(|limit| *limit == 0);
}
- // Column-slice pruning must keep the files holding the filter's
columns.
+ // Derive partition pruning only after excluding masked predicates.
+ if scan.partition_filter.is_none()
+ && !scan.data_predicates.is_empty()
+ && !self.table.schema().partition_keys().is_empty()
+ {
+ let (partition, data) =
super::bucket_filter::split_partition_and_data_predicates(
+ Predicate::and(std::mem::take(&mut scan.data_predicates)),
+ fields,
+ self.table.schema().partition_keys(),
+ );
+ scan.partition_filter = partition.map(|predicate| {
+ PartitionFilter::from_predicate(predicate,
&self.table.schema().partition_fields())
+ });
+ scan.data_predicates = data;
+ }
+ // Keep files for every column used by the rules.
if let Some(ids) = scan.projected_read_field_ids.as_mut() {
- let fields = self.table.schema().fields();
+ let mut columns = rules.filter_columns();
+ for mask in &rules.masks {
+ columns.insert(mask.column);
+ columns.extend(super::query_auth::mask_inputs(mask));
+ }
ids.extend(
- rules
- .filter_columns()
+ columns
.into_iter()
.filter_map(|i| fields.get(i))
.map(|f| f.id()),
@@ -1942,6 +1977,35 @@ impl<'a> PaimonTableScan<'a> {
.table
.authorize_read(query_auth, self.query_auth_select())
.await?;
+ if let Some(grant) = &grant {
+ let fields = self.table.schema().fields();
+ let masked: Vec<&str> = grant
+ .rules()
+ .masks
+ .iter()
+ .map(|m| fields[m.column].name())
+ .collect();
+ if self.chunk_shuffle().is_some()
+ && super::query_auth::leaf_names(&self.data_predicates)
+ .iter()
+ .any(|name| masked.contains(&name.as_str()))
+ {
+ return Err(super::query_auth::unsupported(
+ "chunk_shuffle only supports unmasked partition
predicates",
+ ));
+ }
+ // Explicit partition selectors cannot be reapplied after masking.
+ if let Some(key) = self
+ .partition_filter_columns()
+ .into_iter()
+ .find(|key| masked.contains(&key.as_str()))
+ {
+ return Err(super::query_auth::unsupported(&format!(
+ "the partition key '{key}' is masked, so a filter on it
would match the raw \
+ value"
+ )));
+ }
+ }
Ok(grant)
}
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 36c1a35f..c6882fb5 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -2931,63 +2931,71 @@ async fn test_query_auth_unrestricted_user_can_read() {
);
}
+#[cfg(not(windows))]
+fn blob_payloads() -> (tempfile::TempDir, Vec<u8>, Vec<u8>) {
+ let source = tempfile::tempdir().unwrap();
+ std::fs::write(source.path().join("good.blob"), b"good").unwrap();
+ let descriptor = |file| {
+ paimon::spec::BlobDescriptor::new(
+ format!("file://{}/{file}", source.path().display()),
+ 0,
+ 4,
+ )
+ .serialize()
+ };
+ let (good, missing) = (descriptor("good.blob"),
descriptor("missing.blob"));
+ (source, good, missing)
+}
+
+#[cfg(not(windows))]
+async fn blob_mask_rows(primary_key: bool, payloads: Vec<Option<&[u8]>>) ->
Guarded {
+ let labels = StringArray::from_iter_values((1..=payloads.len()).map(|row|
format!("row{row}")));
+ let batch = RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from_iter_values(1..=payloads.len() as i32))
+ as arrow_array::ArrayRef,
+ ),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from(payloads)) as
arrow_array::ArrayRef,
+ ),
+ ("label", Arc::new(labels) as arrow_array::ArrayRef),
+ ])
+ .unwrap();
+ written(
+ "auth_blob_order",
+ |options| {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::with_nullable(false)))
+ .column("payload", DataType::Blob(BlobType::new()))
+ .column("label",
DataType::VarChar(VarCharType::new(255).unwrap()))
+ .option("blob-descriptor-field", "payload");
+ if primary_key {
+ builder = builder.primary_key(["id"]).option("bucket", "1");
+ } else {
+ builder = builder
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ }
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().unwrap()
+ },
+ vec![batch],
+ )
+ .await
+}
+
/// Authorization must select rows before output or predicate BLOB payloads
open.
#[cfg(not(windows))]
#[tokio::test]
async fn test_query_auth_filters_before_resolving_blob_payloads() {
- let source = tempfile::tempdir().unwrap();
- let good_uri = format!("file://{}/allowed.blob", source.path().display());
- std::fs::write(source.path().join("allowed.blob"), b"good").unwrap();
- let good = paimon::spec::BlobDescriptor::new(good_uri, 0, 4).serialize();
- let missing = paimon::spec::BlobDescriptor::new(
- format!("file://{}/denied.blob", source.path().display()),
- 0,
- 4,
- )
- .serialize();
+ let (_source, good, missing) = blob_payloads();
for primary_key in [true, false] {
- 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![
- None,
- Some(missing.as_slice()),
- Some(good.as_slice()),
- ])) as arrow_array::ArrayRef,
- ),
- ])
- .unwrap();
- let g = written(
- "auth_blob_order",
- |options| {
- let mut builder = Schema::builder()
- .column("id", DataType::Int(IntType::with_nullable(false)))
- .column("payload", DataType::Blob(BlobType::new()))
- .option("blob-descriptor-field", "payload");
- if primary_key {
- builder = builder.primary_key(["id"]).option("bucket",
"1");
- } else {
- builder = builder
- .option("data-evolution.enabled", "true")
- .option("row-tracking.enabled", "true");
- }
- for (key, value) in options {
- builder = builder.option(*key, *value);
- }
- builder.build().unwrap()
- },
- vec![batch],
- )
- .await;
- g.ctx.server.set_auth_response(
- "default",
- "auth_blob_order",
- rules(&[int_leaf(0, "id", "NOT_EQUAL", 2)], &[]),
- );
+ let g = blob_mask_rows(primary_key, vec![None, Some(&missing),
Some(&good)]).await;
+ g.set_rules(&[int_leaf(0, "id", "NOT_EQUAL", 2)], &[]);
let mut id_only = g.table.new_read_builder();
id_only.with_projection(&["id"]).unwrap();
@@ -3040,6 +3048,33 @@ async fn
test_query_auth_filters_before_resolving_blob_payloads() {
}
}
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_query_auth_refuses_blob_masks() {
+ for primary_key in [true, false] {
+ let g = blob_mask_rows(primary_key, vec![None]).await;
+ 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"}),
+ ] {
+ g.set_rules(&[], &[("payload", mask)]);
+ for projection in [vec!["payload"], vec!["id"], vec![]] {
+ 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:?}"
+ );
+ }
+ }
+ }
+}
+
fn schema_of(columns: &[&str], options: &[(&str, &str)]) -> Schema {
let mut builder = Schema::builder();
for name in columns {
@@ -3060,6 +3095,16 @@ struct Guarded {
_tmp: tempfile::TempDir,
}
+impl Guarded {
+ fn set_rules(&self, filters: &[serde_json::Value], masks: &[(&str,
serde_json::Value)]) {
+ self.ctx.server.set_auth_response(
+ self.identifier.database(),
+ self.identifier.object(),
+ rules(filters, masks),
+ );
+ }
+}
+
async fn guarded(name: &str, columns: &[&str]) -> Guarded {
let ctx = setup_catalog(vec!["default"]).await;
let tmp = tempfile::tempdir().unwrap();
@@ -3132,6 +3177,20 @@ fn int_leaf(index: usize, column: &str, function: &str,
literal: i32) -> serde_j
})
}
+fn field_ref_mask(index: usize, column: &str, data_type: &str) ->
serde_json::Value {
+ serde_json::json!({"name": "FIELD_REF", "fieldRef": {
+ "index": index, "name": column, "type": data_type,
+ }})
+}
+
+fn string_ref(index: usize, column: &str) -> serde_json::Value {
+ serde_json::json!({"index": index, "name": column, "type": "VARCHAR(255)"})
+}
+
+fn upper(index: usize, column: &str) -> serde_json::Value {
+ serde_json::json!({"name": "UPPER", "inputs": [string_ref(index, column)]})
+}
+
#[tokio::test]
async fn
test_query_auth_rules_this_client_cannot_parse_fail_closed_at_planning() {
let g = guarded("unparseable", &["id"]).await;
@@ -3558,11 +3617,7 @@ async fn
test_query_auth_row_filter_returns_only_the_rows_it_admits() {
vec![people_rows(1..=5), people_rows(6..=10)],
)
.await;
- g.ctx.server.set_auth_response(
- "default",
- "filtered",
- rules(&[int_leaf(0, "id", "GREATER_THAN", 6)], &[]),
- );
+ g.set_rules(&[int_leaf(0, "id", "GREATER_THAN", 6)], &[]);
let rows = id_names(&read_all(&g.table.new_read_builder()).await.unwrap());
let expected: Vec<_> = (7..=10)
@@ -3620,15 +3675,9 @@ async fn
test_query_auth_refuses_an_engine_decoder_filter() {
let mut id_only = g.table.new_read_builder();
id_only.with_projection(&["id"]).unwrap();
// Under a row filter, and under a column permission alone.
- g.ctx.server.set_auth_response(
- "default",
- "decoded",
- rules(&[int_leaf(0, "id", "GREATER_THAN", 6)], &[]),
- );
+ g.set_rules(&[int_leaf(0, "id", "GREATER_THAN", 6)], &[]);
let under_rules = id_only.new_scan().plan().await.unwrap();
- g.ctx
- .server
- .set_auth_response("default", "decoded", rules(&[], &[]));
+ g.set_rules(&[], &[]);
g.ctx
.server
.set_column_auth("default", "decoded", vec!["id".to_string()]);
@@ -3651,6 +3700,660 @@ async fn
test_query_auth_refuses_an_engine_decoder_filter() {
}
}
+#[tokio::test]
+async fn test_query_auth_refuses_a_mask_that_reads_another_masked_column() {
+ let g = guarded("chained", &["id", "a", "b"]).await;
+ let field_ref = |index, column: &str| field_ref_mask(index, column, "INT");
+ let hidden = serde_json::json!({"name": "NULL"});
+
+ g.set_rules(&[], &[("a", hidden.clone()), ("b", field_ref(0, "id"))]);
+ g.table
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .expect("a mask may read an unmasked column");
+
+ g.set_rules(&[], &[("a", hidden), ("b", field_ref(1, "a"))]);
+ assert_refused(plan_err(&g.table, "'b' would publish the raw value of
'a'").await);
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_query_auth_widens_an_integer_mask() {
+ let schema = |options: &[(&str, &str)]| {
+ let mut builder = Schema::builder()
+ .column("input", DataType::Int(IntType::new()))
+ .column("wide", DataType::BigInt(BigIntType::new()));
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().unwrap()
+ };
+ let batch = RecordBatch::try_from_iter(vec![
+ (
+ "input",
+ Arc::new(Int32Array::from(vec![Some(i32::MIN), None,
Some(i32::MAX)]))
+ as Arc<dyn Array>,
+ ),
+ (
+ "wide",
+ Arc::new(Int64Array::from(vec![999; 3])) as Arc<dyn Array>,
+ ),
+ ])
+ .unwrap();
+ let g = written("widened", schema, vec![batch]).await;
+ g.set_rules(
+ &[],
+ &[(
+ "wide",
+ serde_json::json!({
+ "name": "CAST", "fieldRef": {"index": 0, "name": "input",
"type": "INT"},
+ "type": "BIGINT",
+ }),
+ )],
+ );
+ let mut builder = g.table.new_read_builder();
+ builder.with_projection(&["wide"]).unwrap();
+ let batches = read_all(&builder).await.unwrap();
+ assert!(batches.iter().all(|batch| batch.num_columns() == 1));
+ let values: Vec<_> = batches
+ .iter()
+ .flat_map(|batch| column::<Int64Array>(batch, "wide").iter())
+ .collect();
+ assert_eq!(
+ values,
+ vec![Some(i64::from(i32::MIN)), None, Some(i64::from(i32::MAX))]
+ );
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_query_auth_filters_and_masks_the_same_column() {
+ for (primary_key, evolution, blob) in [
+ (false, false, false),
+ (true, false, false),
+ (false, true, false),
+ (true, false, true),
+ (false, true, true),
+ ] {
+ let schema = |options: &[(&str, &str)]| {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name",
DataType::VarChar(VarCharType::new(255).unwrap()));
+ if primary_key {
+ builder = builder.primary_key(["id"]).option("bucket", "1");
+ }
+ if evolution {
+ builder = builder
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ }
+ if blob {
+ builder = builder.column("payload",
DataType::Blob(BlobType::new()));
+ }
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().unwrap()
+ };
+ let mut columns = vec![
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4])) as Arc<dyn Array>,
+ ),
+ (
+ "name",
+ Arc::new(StringArray::from(vec![
+ Some("alice"),
+ Some("ALICE"),
+ Some("alice"),
+ None,
+ ])) as Arc<dyn Array>,
+ ),
+ ];
+ if blob {
+ columns.push((
+ "payload",
+ Arc::new(LargeBinaryArray::from(vec![None::<&[u8]>; 4])),
+ ));
+ }
+ let g = written(
+ "same_column",
+ schema,
+ vec![RecordBatch::try_from_iter(columns).unwrap()],
+ )
+ .await;
+ g.set_rules(
+ &[serde_json::json!({
+ "kind": "LEAF", "function": "EQUAL", "literals": ["alice"],
+ "transform": {"name": "FIELD_REF", "fieldRef": string_ref(1,
"name")},
+ })],
+ &[("name", upper(1, "name"))],
+ );
+ assert_eq!(
+ id_names(&read_all(&g.table.new_read_builder()).await.unwrap()),
+ vec![
+ (1, Some("ALICE".to_string())),
+ (3, Some("ALICE".to_string()))
+ ],
+ );
+ let predicates = PredicateBuilder::new(g.table.schema().fields());
+ for (value, expected) in [("ALICE", vec![3]), ("alice", vec![])] {
+ let mut builder = g.table.new_read_builder();
+ builder.with_filter(Predicate::And(vec![
+ predicates.greater_than("id", Datum::Int(1)).unwrap(),
+ predicates
+ .equal("name", Datum::String(value.to_string()))
+ .unwrap(),
+ ]));
+ builder.with_limit(1);
+ if !blob {
+ builder.with_projection(&["id"]).unwrap();
+ }
+ assert_eq!(ids(&read_all(&builder).await.unwrap()), expected);
+ }
+ }
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_query_auth_keeps_a_mask_source_in_another_partial_file() {
+ let schema = |options: &[(&str, &str)]| {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("visible",
DataType::VarChar(VarCharType::new(255).unwrap()))
+ .column("source",
DataType::VarChar(VarCharType::new(255).unwrap()))
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().unwrap()
+ };
+ let g = written("partial_mask", schema, vec![]).await;
+ let mut options = Options::new();
+ options.set(
+ CatalogOptions::WAREHOUSE,
+ format!("file://{}", g._tmp.path().display()),
+ );
+ let catalog = FileSystemCatalog::new(options).unwrap();
+ let plain = catalog.get_table(&g.identifier).await.unwrap();
+ let writer = plain.new_write_builder();
+ let mut write = writer.new_write().unwrap();
+ write
+ .with_write_type(vec!["id".into(), "visible".into()])
+ .unwrap();
+ write
+ .write_arrow_batch(
+ &RecordBatch::try_from_iter(vec![
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![1, 2])) as Arc<dyn Array>,
+ ),
+ (
+ "visible",
+ Arc::new(StringArray::from(vec!["raw1", "raw2"])) as
Arc<dyn Array>,
+ ),
+ ])
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ writer
+ .new_commit()
+ .commit(write.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ let update = RecordBatch::try_from_iter(vec![
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from(vec![0, 1])) as Arc<dyn Array>,
+ ),
+ (
+ "source",
+ Arc::new(StringArray::from(vec!["alice", "bob"])) as Arc<dyn
Array>,
+ ),
+ ])
+ .unwrap();
+ let messages = writer
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![update])
+ .await
+ .unwrap();
+ writer.new_commit().commit(messages).await.unwrap();
+
+ let plan = plain.new_read_builder().new_scan().plan().await.unwrap();
+ let files: Vec<_> = plan
+ .splits()
+ .iter()
+ .flat_map(|split| split.data_files())
+ .collect();
+ assert_eq!(files.len(), 2);
+ assert!(files
+ .iter()
+ .any(|file| file.write_cols == Some(vec!["source".to_string()])));
+ let mut projected = plain.new_read_builder();
+ projected.with_projection(&["visible"]).unwrap();
+ let plan = projected.new_scan().plan().await.unwrap();
+ assert_eq!(
+ plan.splits()
+ .iter()
+ .map(|split| split.data_files().len())
+ .sum::<usize>(),
+ 1
+ );
+ g.set_rules(
+ &[],
+ &[("visible", field_ref_mask(2, "source", "VARCHAR(255)"))],
+ );
+ for filtered in [false, true] {
+ let mut builder = g.table.new_read_builder();
+ builder.with_projection(&["visible"]).unwrap();
+ if filtered {
+ builder.with_filter(
+ PredicateBuilder::new(g.table.schema().fields())
+ .equal("visible", Datum::String("bob".to_string()))
+ .unwrap(),
+ );
+ builder.with_limit(1);
+ }
+ let batches = read_all(&builder).await.unwrap();
+ assert!(batches.iter().all(|batch| batch.num_columns() == 1));
+ let values: Vec<_> = batches
+ .iter()
+ .flat_map(|batch| column::<StringArray>(batch, "visible").iter())
+ .collect();
+ assert_eq!(
+ values,
+ if filtered {
+ vec![Some("bob")]
+ } else {
+ vec![Some("alice"), Some("bob")]
+ }
+ );
+ }
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_query_auth_masks_a_column_and_leaves_the_others_raw() {
+ let g = written(
+ "masked",
+ people,
+ vec![people_batch(&[(1, "alice"), (2, "carol")])],
+ )
+ .await;
+ let name = || string_ref(1, "name");
+ for (mask, expected) in [
+ (serde_json::json!({"name": "NULL"}), [None, None]),
+ (upper(1, "name"), [Some("ALICE"), Some("CAROL")]),
+ // Arrow strings do not enforce VARCHAR width.
+ (
+ serde_json::json!({"name": "CAST", "fieldRef": name(), "type":
"VARCHAR(3)"}),
+ [Some("ali"), Some("car")],
+ ),
+ (
+ serde_json::json!({"name": "CONCAT", "inputs": ["user-", name()]}),
+ [Some("user-alice"), Some("user-carol")],
+ ),
+ ] {
+ g.set_rules(&[], &[("name", mask.clone())]);
+ assert_eq!(
+ id_names(&read_all(&g.table.new_read_builder()).await.unwrap()),
+ vec![
+ (1, expected[0].map(String::from)),
+ (2, expected[1].map(String::from))
+ ],
+ "{mask}"
+ );
+ }
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_query_auth_filters_a_masked_column_on_its_masked_value() {
+ let g = written(
+ "upper",
+ people,
+ vec![people_batch(&[(1, "alice"), (2, "bob")])],
+ )
+ .await;
+ g.set_rules(&[], &[("name", upper(1, "name"))]);
+ let name_is = |value: &str| {
+ PredicateBuilder::new(g.table.schema().fields())
+ .equal("name", Datum::String(value.to_string()))
+ .unwrap()
+ };
+
+ // Raw bounds exclude 'ALICE'; masked predicates must not prune on them.
+ let mut masked = g.table.new_read_builder();
+ masked.with_filter(name_is("ALICE"));
+ assert_eq!(
+ id_names(&read_all(&masked).await.unwrap()),
+ vec![(1, Some("ALICE".to_string()))]
+ );
+ masked.with_projection(&["id"]).unwrap();
+ assert_eq!(ids(&read_all(&masked).await.unwrap()), vec![1]);
+
+ let mut raw = g.table.new_read_builder();
+ raw.with_filter(name_is("alice"));
+ assert!(read_all(&raw)
+ .await
+ .unwrap()
+ .iter()
+ .all(|b| b.num_rows() == 0));
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_query_auth_does_not_prune_buckets_on_a_masked_bucket_key() {
+ // Bucket index 0 names `name`, while schema index 0 names `id`.
+ let schema = |options: &[(&str, &str)]| {
+ let mut all = vec![("bucket", "4"), ("bucket-key", "name")];
+ all.extend_from_slice(options);
+ people(&all)
+ };
+ let g = written("bucketed", schema, vec![people_rows(1..=10)]).await;
+ g.set_rules(&[], &[("name", upper(1, "name"))]);
+ for (id, name) in (1..=10).zip(NAMES) {
+ let mut builder = g.table.new_read_builder();
+ builder.with_filter(
+ PredicateBuilder::new(g.table.schema().fields())
+ .equal("name", Datum::String(name.to_uppercase()))
+ .unwrap(),
+ );
+ assert_eq!(
+ id_names(&read_all(&builder).await.unwrap()),
+ vec![(id, Some(name.to_uppercase()))],
+ "the raw value's bucket is not the masked value's"
+ );
+ }
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_query_auth_chunk_shuffle_with_partition_filters() {
+ for evolution in [false, true] {
+ let schema = |options: &[(&str, &str)]| {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name",
DataType::VarChar(VarCharType::new(255).unwrap()))
+ .partition_keys(["name"]);
+ if evolution {
+ builder = builder
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ }
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().unwrap()
+ };
+ let g = written(
+ "partition_shuffle",
+ schema,
+ vec![people_batch(&[
+ (1, "eng"),
+ (2, "eng"),
+ (3, "ops"),
+ (4, "eng"),
+ (5, "ops"),
+ (6, "eng"),
+ ])],
+ )
+ .await;
+ let predicates = PredicateBuilder::new(g.table.schema().fields());
+ let partition = predicates
+ .equal("name", Datum::String("eng".into()))
+ .unwrap();
+ for (filters, masks, expected) in [
+ (vec![], vec![], vec![1, 2, 4, 6]),
+ (
+ vec![],
+ vec![("id", field_ref_mask(0, "id", "INT"))],
+ vec![1, 2, 4, 6],
+ ),
+ (
+ vec![int_leaf(0, "id", "GREATER_THAN", 1)],
+ vec![],
+ vec![2, 4, 6],
+ ),
+ ] {
+ g.set_rules(&filters, &masks);
+ let mut read = g.table.new_read_builder();
+ read.with_projection(&["id"])
+ .unwrap()
+ .with_filter(partition.clone());
+ let scan = read
+ .new_scan()
+ .with_chunk_shuffle(7, 2)
+ .expect("partition-only filters must allow chunk shuffle");
+ for traced in [false, true] {
+ let plan = if traced {
+ scan.plan_with_trace().await.unwrap().0
+ } else {
+ scan.plan().await.unwrap()
+ };
+ assert_eq!(plan.splits().len(), 2);
+ assert!(plan
+ .splits()
+ .iter()
+ .all(|split| split.partition().get_string(0).unwrap() ==
"eng"));
+ let batches: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ ids(&batches),
+ expected,
+ "evolution={evolution}, traced={traced}"
+ );
+ }
+ }
+
+ let data = predicates.greater_than("id", Datum::Int(2)).unwrap();
+ for filter in [
+ data.clone(),
+ Predicate::and(vec![partition.clone(), data.clone()]),
+ Predicate::or(vec![partition.clone(), data]),
+ ] {
+ let mut read = g.table.new_read_builder();
+ read.with_filter(filter);
+ let error = read.new_scan().with_chunk_shuffle(7, 2).unwrap_err();
+ assert!(
+ matches!(error, paimon::Error::Unsupported { ref message }
+ if message.contains("only supports partition predicates")),
+ "{error:?}"
+ );
+ }
+
+ g.set_rules(&[], &[("name", upper(1, "name"))]);
+ let mut read = g.table.new_read_builder();
+ read.with_filter(
+ predicates
+ .equal("name", Datum::String("ENG".into()))
+ .unwrap(),
+ );
+ let scan = read.new_scan().with_chunk_shuffle(7, 2).unwrap();
+ for traced in [false, true] {
+ let result = if traced {
+ scan.plan_with_trace().await.map(|(plan, _)| plan)
+ } else {
+ scan.plan().await
+ };
+ let error = result.unwrap_err();
+ assert!(
+ matches!(error, paimon::Error::Unsupported { ref message }
+ if message.contains("unmasked partition predicates")),
+ "{error:?}"
+ );
+ }
+ }
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_query_auth_filters_masked_partition_values() {
+ for data_evolution in [false, true] {
+ let schema = |options: &[(&str, &str)]| {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("dept",
DataType::VarChar(VarCharType::new(255).unwrap()))
+ .column("src",
DataType::VarChar(VarCharType::new(255).unwrap()))
+ .column("region",
DataType::VarChar(VarCharType::new(255).unwrap()))
+ .partition_keys(["dept", "region"]);
+ if data_evolution {
+ builder = builder
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ }
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().unwrap()
+ };
+ let batch = RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4])) as
arrow_array::ArrayRef,
+ ),
+ (
+ "dept",
+ Arc::new(StringArray::from(vec![
+ Some("eng"),
+ Some("ops"),
+ Some("eng"),
+ None,
+ ])) as arrow_array::ArrayRef,
+ ),
+ (
+ "src",
+ Arc::new(StringArray::from(vec![
+ Some("OPS"),
+ Some("ENG"),
+ Some("HR"),
+ None,
+ ])) as arrow_array::ArrayRef,
+ ),
+ (
+ "region",
+ Arc::new(StringArray::from(vec!["east", "east", "west",
"east"]))
+ as arrow_array::ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let g = written("by_dept", schema, vec![batch]).await;
+ let predicates = PredicateBuilder::new(g.table.schema().fields());
+ let equals = |column: &str, value: &str| {
+ predicates
+ .equal(column, Datum::String(value.into()))
+ .unwrap()
+ };
+ for (mask, expected, nulls) in [
+ (upper(1, "dept"), vec![1, 3], vec![4]),
+ (field_ref_mask(2, "src", "VARCHAR(255)"), vec![2], vec![4]),
+ (
+ serde_json::json!({"name": "NULL"}),
+ vec![],
+ vec![1, 2, 3, 4],
+ ),
+ ] {
+ g.set_rules(&[], &[("dept", mask)]);
+ for projection in [vec!["id", "dept"], vec!["id"]] {
+ let mut read = g.table.new_read_builder();
+ read.with_projection(&projection)
+ .unwrap()
+ .with_filter(equals("dept", "ENG"));
+ let batches = read_all(&read).await.unwrap();
+ assert_eq!(ids(&batches), expected);
+ if projection.len() == 2 {
+ for batch in &batches {
+ assert!(column::<StringArray>(batch, "dept")
+ .iter()
+ .all(|value| value == Some("ENG")));
+ }
+ }
+ }
+ let mut disjunction = expected.clone();
+ disjunction.push(3);
+ disjunction.sort_unstable();
+ disjunction.dedup();
+ for (predicate, expected) in [
+ (equals("dept", "eng"), vec![]),
+ (predicates.is_null("dept").unwrap(), nulls),
+ (
+ paimon::spec::Predicate::or(vec![
+ equals("dept", "ENG"),
+ equals("region", "west"),
+ ]),
+ disjunction,
+ ),
+ ] {
+ let mut read = g.table.new_read_builder();
+ read.with_projection(&["id"])
+ .unwrap()
+ .with_filter(predicate);
+ assert_eq!(ids(&read_all(&read).await.unwrap()), expected);
+ }
+
+ let mut limited = g.table.new_read_builder();
+ limited
+ .with_projection(&["id"])
+ .unwrap()
+ .with_filter(paimon::spec::Predicate::and(vec![
+ equals("dept", "ENG"),
+ equals("region", "east"),
+ ]))
+ .with_limit(1);
+ let (plan, trace) =
limited.new_scan().plan_with_trace().await.unwrap();
+ assert_eq!(trace.limit, None);
+ assert!(plan
+ .splits()
+ .iter()
+ .all(|split| split.partition().get_string(1).unwrap() ==
"east"));
+ let rows: Vec<RecordBatch> = limited
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let expected: Vec<_> = expected.into_iter().filter(|id| *id !=
3).take(1).collect();
+ assert_eq!(ids(&rows), expected);
+ limited.with_limit(0);
+ assert!(ids(&read_all(&limited).await.unwrap()).is_empty());
+ }
+
+ // Unrestricted users still get raw partition pruning and its limit.
+ g.set_rules(&[], &[]);
+ let mut plain = g.table.new_read_builder();
+ plain
+ .with_projection(&["id"])
+ .unwrap()
+ .with_filter(equals("dept", "eng"));
+ assert_eq!(ids(&read_all(&plain).await.unwrap()), vec![1, 3]);
+ plain.with_limit(1);
+ let (plan, trace) = plain.new_scan().plan_with_trace().await.unwrap();
+ assert_eq!(trace.limit, Some(1));
+ assert!(!plan.splits().is_empty());
+ let rows = ids(&read_all(&plain).await.unwrap());
+ assert_eq!(rows.len(), 1);
+ assert!([1, 3].contains(&rows[0]));
+
+ g.ctx
+ .server
+ .set_column_auth("default", "by_dept", vec!["id".to_string()]);
+ assert_forbidden(read_all(&plain).await.unwrap_err());
+ }
+}
+
#[cfg(not(windows))]
#[tokio::test]
async fn
test_query_auth_asks_about_the_projection_and_the_columns_filters_read() {
@@ -3712,15 +4415,9 @@ async fn
test_query_auth_refuses_a_read_wider_than_its_plan() {
let wider = [g.table.new_read_builder(), filtered];
// Scoped once under a row filter, once by a column permission alone.
- g.ctx.server.set_auth_response(
- "default",
- "narrow",
- rules(&[int_leaf(0, "id", "GREATER_THAN", 6)], &[]),
- );
+ g.set_rules(&[int_leaf(0, "id", "GREATER_THAN", 6)], &[]);
let under_rules = planned.new_scan().plan().await.unwrap();
- g.ctx
- .server
- .set_auth_response("default", "narrow", rules(&[], &[]));
+ g.set_rules(&[], &[]);
g.ctx
.server
.set_column_auth("default", "narrow", vec!["id".to_string()]);
@@ -3904,11 +4601,7 @@ fn score_rows(batches: &[RecordBatch]) -> Vec<(i32,
String, i32)> {
#[tokio::test]
async fn test_query_auth_filters_the_merged_primary_key_row() {
let g = scores().await;
- g.ctx.server.set_auth_response(
- "default",
- "scores",
- rules(&[int_leaf(2, "score", "GREATER_THAN", 30)], &[]),
- );
+ g.set_rules(&[int_leaf(2, "score", "GREATER_THAN", 30)], &[]);
// Filtering each commit before the merge would bring back bob's first row.
assert_eq!(
score_rows(&read_all(&g.table.new_read_builder()).await.unwrap()),
@@ -3916,6 +4609,20 @@ async fn
test_query_auth_filters_the_merged_primary_key_row() {
);
}
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_query_auth_masks_the_merged_primary_key_row() {
+ let g = scores().await;
+ g.set_rules(
+ &[int_leaf(2, "score", "GREATER_THAN", 30)],
+ &[("name", upper(1, "name"))],
+ );
+ assert_eq!(
+ score_rows(&read_all(&g.table.new_read_builder()).await.unwrap()),
+ vec![(1, "ALICIA".to_string(), 80), (3, "CAROL".to_string(), 90)]
+ );
+}
+
#[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 5bcd622f..b9a2e64f 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1646,13 +1646,14 @@ nullness can differ from its upstream value. Other BLOB
payloads and output-only
BLOB views are resolved after row authorization.
When a table in a REST catalog has `'query-auth.enabled' = 'true'`, planning a
-query asks the catalog what the current user may read, and the row filters it
-returns (see `create_policy`) are applied to the result:
-
-- Rows are filtered on their stored values.
-- A user the catalog also masks columns for is refused.
-- Statistics are not used under a row filter, which also turns off `LIMIT`
- pushdown, so such queries read the data.
+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.
+- 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.
- Time travel, branches, system tables, incremental and audit-log reads, and
search are refused on such tables.
