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 4ceef6be [read] Support transformed REST query authorization filters
(#1124)
4ceef6be is described below
commit 4ceef6be00b2df19eaef0fa63c5ea3b82660fcf6
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Oct 14 08:47:16 2026 +0800
[read] Support transformed REST query authorization filters (#1124)
---
crates/paimon/src/spec/types.rs | 6 +
crates/paimon/src/table/query_auth.rs | 2 +
crates/paimon/src/table/query_auth/row_filter.rs | 620 +++++++++++++++++++++++
crates/paimon/src/table/query_auth/rules.rs | 169 +++---
crates/paimon/src/table/source.rs | 2 +-
crates/paimon/src/table/table_read.rs | 7 +-
crates/paimon/tests/rest_catalog_test.rs | 80 +++
7 files changed, 820 insertions(+), 66 deletions(-)
diff --git a/crates/paimon/src/spec/types.rs b/crates/paimon/src/spec/types.rs
index ab73da7e..0ff00663 100644
--- a/crates/paimon/src/spec/types.rs
+++ b/crates/paimon/src/spec/types.rs
@@ -887,6 +887,11 @@ impl FromStr for CharType {
.fail();
}
+ // Java DataTypeJsonParser defaults a bare CHAR to length one.
+ if s == "CHAR" || s == "CHAR NOT NULL" {
+ return Self::with_nullable(s == "CHAR", Self::DEFAULT_LENGTH);
+ }
+
let (open_bracket, close_bracket) =
serde_utils::extract_brackets_pos(s, "CharType")?;
let length_str = &s[open_bracket + 1..close_bracket];
let length = length_str
@@ -2897,6 +2902,7 @@ mod tests {
fn test_default_length_string_types_match_java() {
// Java DataTypeJsonParser uses each type's DEFAULT_LENGTH when
omitted.
let cases = [
+ ("CHAR", DataType::Char(CharType::default())),
("VARCHAR", DataType::VarChar(VarCharType::default())),
("BINARY", DataType::Binary(BinaryType::default())),
("VARBINARY", DataType::VarBinary(VarBinaryType::default())),
diff --git a/crates/paimon/src/table/query_auth.rs
b/crates/paimon/src/table/query_auth.rs
index 3de5625a..74666129 100644
--- a/crates/paimon/src/table/query_auth.rs
+++ b/crates/paimon/src/table/query_auth.rs
@@ -17,6 +17,8 @@
//! What the REST server authorized a user to read from one table.
+mod row_filter;
+pub(crate) use row_filter::RowFilter;
mod rules;
use crate::api::AuthTableQueryResponse;
diff --git a/crates/paimon/src/table/query_auth/row_filter.rs
b/crates/paimon/src/table/query_auth/row_filter.rs
new file mode 100644
index 00000000..7c8afec6
--- /dev/null
+++ b/crates/paimon/src/table/query_auth/row_filter.rs
@@ -0,0 +1,620 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! REST row predicates evaluated on transform outputs, as Java LeafPredicate
does.
+
+use super::rules::{
+ canonicalize_nan, check_transform_cast, column_of, evaluate_transform,
rule_mask,
+ transform_output_type,
+};
+use super::unsupported;
+use crate::spec::{DataField, DataType, Predicate, Transform};
+use crate::Result;
+use arrow_array::{ArrayRef, BooleanArray, RecordBatch};
+use std::collections::HashSet;
+
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) enum RowFilter {
+ /// Existing field comparisons keep the ordinary predicate representation.
+ Predicate(Predicate),
+ /// The predicate compares the transform's result, not its source column.
+ Transformed {
+ transform: Transform,
+ predicate: Predicate,
+ data_type: DataType,
+ },
+ And(Vec<RowFilter>),
+ Or(Vec<RowFilter>),
+}
+
+impl RowFilter {
+ pub(super) fn parse(json: &str, fields: &[DataField]) -> Result<Self> {
+ let value = serde_json::from_str(json)
+ .map_err(|e| unsupported(&format!("invalid filter JSON: {e}")))?;
+ Self::parse_value(&value, fields)
+ }
+
+ fn parse_value(value: &serde_json::Value, fields: &[DataField]) ->
Result<Self> {
+ let mut normalized = value.clone();
+ if normalized["kind"] == "LEAF"
+ && normalized["literals"].is_null()
+ && matches!(
+ normalized["function"].as_str(),
+ Some("TRUE" | "FALSE" | "IS_NULL" | "IS_NOT_NULL")
+ )
+ {
+ normalized["literals"] = serde_json::json!([]);
+ }
+ let value = &normalized;
+ match value["kind"].as_str() {
+ Some("COMPOUND") => {
+ let children = value["children"]
+ .as_array()
+ .filter(|c| !c.is_empty())
+ .ok_or_else(|| unsupported("the server's row filter has
empty children"))?;
+ let children = children
+ .iter()
+ .map(|child| Self::parse_value(child, fields))
+ .collect::<Result<Vec<_>>>()?;
+ match value["function"].as_str() {
+ Some("AND") => Ok(Self::And(children)),
+ Some("OR") => Ok(Self::Or(children)),
+ _ => Err(unsupported("unknown compound filter function")),
+ }
+ }
+ Some("LEAF") => {
+ let transform =
Transform::from_rest_json(&value["transform"].to_string(), fields)?;
+ let data_type = transform_output_type(&transform, fields)?;
+ if matches!(data_type, DataType::Binary(_) |
DataType::VarBinary(_))
+ && value["literals"]
+ .as_array()
+ .is_some_and(|literals| literals.iter().any(|literal|
!literal.is_null()))
+ {
+ return Err(unsupported(
+ "Java predicates do not support non-null binary
literals",
+ ));
+ }
+ if matches!(transform, Transform::FieldRef(_))
+ && !matches!(value["function"].as_str(), Some("TRUE" |
"FALSE"))
+ {
+ return Predicate::from_rest_json(&value.to_string(),
fields)
+ .map(Self::Predicate);
+ }
+ check_transform_cast(&transform, fields)?;
+ if let Transform::Cast(index, _) = &transform {
+ // Java preserves the original cast executor when
rebinding.
+ // Do not silently choose a different cast for an evolved
input.
+ let stored: DataType =
+
serde_json::from_value(value["transform"]["fieldRef"]["type"].clone())
+ .map_err(|e| unsupported(&format!("invalid CAST
source type: {e}")))?;
+ if stored.copy_with_nullable(true)?
+ != fields[*index].data_type().copy_with_nullable(true)?
+ {
+ return Err(unsupported(
+ "CAST with a rebound source type requires its
original Java executor",
+ ));
+ }
+ }
+
+ let output = vec![DataField::new(0, "value".into(),
data_type.clone())];
+ let mut leaf = value.clone();
+ // Reuse the existing typed literal/function parser on the
expression
+ // output. This synthetic field is never a physical table
column.
+ leaf["transform"] = serde_json::json!({
+ "name": "FIELD_REF", "fieldRef": {"name": "value"}
+ });
+ let predicate =
+ if let Some(function @ ("TRUE" | "FALSE")) =
value["function"].as_str() {
+ // Java deserializes every literal using
transform.outputType(),
+ // even when the function's result is constant.
+ leaf["function"] = serde_json::json!("IN");
+ Predicate::from_rest_json(&leaf.to_string(), &output)?;
+ if function == "TRUE" {
+ Predicate::AlwaysTrue
+ } else {
+ Predicate::AlwaysFalse
+ }
+ } else {
+ Predicate::from_rest_json(&leaf.to_string(), &output)?
+ };
+ Ok(Self::Transformed {
+ transform,
+ predicate,
+ data_type,
+ })
+ }
+ _ => Err(unsupported("unknown row predicate kind")),
+ }
+ }
+
+ pub(super) fn collect_field_indices(&self, out: &mut HashSet<usize>) {
+ match self {
+ Self::Predicate(predicate) =>
predicate.collect_leaf_field_indices(out),
+ Self::Transformed { transform, .. } =>
transform.collect_field_indices(out),
+ Self::And(children) | Self::Or(children) => children
+ .iter()
+ .for_each(|child| child.collect_field_indices(out)),
+ }
+ }
+
+ pub(super) fn evaluate(
+ &self,
+ batch: &RecordBatch,
+ fields: &[DataField],
+ batch_fields: &[DataField],
+ ) -> Result<BooleanArray> {
+ match self {
+ Self::Predicate(predicate) => rule_mask(batch, predicate, fields,
batch_fields),
+ Self::Transformed {
+ transform,
+ predicate,
+ data_type,
+ } => {
+ let input = |index: usize| -> Result<ArrayRef> {
+ let field = fields.get(index).ok_or_else(|| {
+ unsupported("the server's transform reads an unknown
column")
+ })?;
+ column_of(batch, field, batch_fields)
+ };
+ // Evaluate before the function, even for FALSE / a NULL
literal:
+ // Java LeafPredicate.test first calls
transform.transform(row).
+ let values = evaluate_transform(
+ transform,
+ batch.num_rows(),
+ &input,
+ &crate::arrow::paimon_type_to_arrow(data_type)?,
+ )?;
+ let field = DataField::new(0, "value".into(),
data_type.clone());
+ let values = canonicalize_nan(&values);
+ let output = RecordBatch::try_from_iter(vec![("value",
values)])
+ .map_err(|e| unsupported(&format!("cannot evaluate filter
transform: {e}")))?;
+ rule_mask(
+ &output,
+ predicate,
+ std::slice::from_ref(&field),
+ std::slice::from_ref(&field),
+ )
+ }
+ Self::And(children) | Self::Or(children) => {
+ let or = matches!(self, Self::Or(_));
+ let mut result = vec![!or; batch.num_rows()];
+ let mut pending: Vec<usize> = (0..batch.num_rows()).collect();
+ for child in children {
+ if pending.is_empty() {
+ break;
+ }
+ // Preserve Java's row-by-row short circuit. A later
substring
+ // or cast may fail on rows already decided by an earlier
child.
+ let active = if pending.len() == batch.num_rows() {
+ batch.clone()
+ } else {
+ let selected =
+ BooleanArray::from(result.iter().map(|v| *v !=
or).collect::<Vec<_>>());
+ arrow_select::filter::filter_record_batch(batch,
&selected)
+ .map_err(|e| unsupported(&format!("cannot select
filter rows: {e}")))?
+ };
+ let mask = child.evaluate(&active, fields, batch_fields)?;
+ let mut undecided = Vec::new();
+ for (row, value) in pending.into_iter().zip(mask.iter()) {
+ let value = value.unwrap_or(false);
+ result[row] = value;
+ if value != or {
+ undecided.push(row);
+ }
+ }
+ pending = undecided;
+ }
+ Ok(BooleanArray::from(result))
+ }
+ }
+ }
+}
+
+impl From<Predicate> for RowFilter {
+ fn from(predicate: Predicate) -> Self {
+ Self::Predicate(predicate)
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::{BigIntType, DateType, IntType, VarCharType};
+ use crate::table::query_auth::rules::filter_batch;
+ use arrow_array::{Date32Array, Int32Array, Int64Array, StringArray};
+ use serde_json::{json, Value};
+ use std::sync::Arc;
+
+ fn fields() -> Vec<DataField> {
+ vec![
+ DataField::new(0, "id".into(), DataType::Int(IntType::new())),
+ DataField::new(1, "s".into(),
DataType::VarChar(VarCharType::string_type())),
+ DataField::new(2, "p".into(), DataType::BigInt(BigIntType::new())),
+ DataField::new(3, "d".into(), DataType::Date(DateType::new())),
+ ]
+ }
+
+ fn rows() -> RecordBatch {
+ RecordBatch::try_from_iter(vec![
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![0, 1, 2, 3])) as ArrayRef,
+ ),
+ (
+ "s",
+ Arc::new(StringArray::from(vec![
+ Some("η²πδΉ"),
+ Some("abcabc"),
+ Some(""),
+ None,
+ ])),
+ ),
+ ("p", Arc::new(Int64Array::from(vec![2, -3, 0, i64::MAX]))),
+ (
+ "d",
+ Arc::new(Date32Array::from(vec![
+ Some(19782),
+ Some(20087),
+ Some(-1),
+ None,
+ ])),
+ ),
+ ])
+ .unwrap()
+ }
+
+ fn input(name: &str, type_: &str) -> Value {
+ json!({"name": name, "type": type_, "index": 99})
+ }
+
+ fn leaf(transform: Value, function: &str, literals: Value) -> Value {
+ json!({"kind": "LEAF", "transform": transform, "function": function,
"literals": literals})
+ }
+
+ fn kept(rule: Value) -> Vec<i32> {
+ let fields = fields();
+ let filter = RowFilter::parse(&rule.to_string(), &fields).unwrap();
+ let result = filter_batch(&rows(), &[filter], &fields,
&fields).unwrap();
+ result
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ }
+
+ #[test]
+ fn all_existing_string_transforms_filter_their_outputs() {
+ let source = input("s", "STRING");
+ let cases = [
+ (
+ json!({"name":"UPPER","inputs":[source]}),
+ "EQUAL",
+ json!(["ABCABC"]),
+ vec![1],
+ ),
+ (
+ json!({"name":"LOWER","inputs":["ABC"]}),
+ "EQUAL",
+ json!(["abc"]),
+ vec![0, 1, 2, 3],
+ ),
+ (
+ json!({"name":"CONCAT","inputs":[source,"x"]}),
+ "EQUAL",
+ json!(["abcabcx"]),
+ vec![1],
+ ),
+ (
+ json!({"name":"CONCAT","inputs":[]}),
+ "EQUAL",
+ json!([""]),
+ vec![0, 1, 2, 3],
+ ),
+ (
+ json!({"name":"CONCAT_WS","inputs":["-",source,null]}),
+ "EQUAL",
+ json!([""]),
+ vec![2, 3],
+ ),
+ (
+ json!({"name":"SUBSTRING","inputs":[source,2,2]}),
+ "EQUAL",
+ json!(["πδΉ"]),
+ vec![0],
+ ),
+ (
+
json!({"name":"SUBSTRING","inputs":[source,input("p","BIGINT"),2]}),
+ "EQUAL",
+ json!(["ab"]),
+ vec![1],
+ ),
+ (
+
json!({"name":"TRIM","trimFlag":"BOTH","inputs":[source,"η²δΉac"]}),
+ "EQUAL",
+ json!(["π"]),
+ vec![0],
+ ),
+ (
+ json!({"name":"LENGTH","inputs":[source]}),
+ "BETWEEN",
+ json!([3, 6]),
+ vec![0, 1],
+ ),
+ (
+ json!({"name":"BIT_LENGTH","inputs":[source]}),
+ "GREATER_THAN",
+ json!([48]),
+ vec![0],
+ ),
+ (
+
json!({"name":"PAD","direction":"LEFT","inputs":[source,5,"πx"]}),
+ "EQUAL",
+ json!(["πxη²πδΉ"]),
+ vec![0],
+ ),
+ (
+ json!({"name":"TRANSLATE","inputs":[source,"η²πδΉaba","XY"]}),
+ "EQUAL",
+ json!(["XY"]),
+ vec![0],
+ ),
+ (
+ json!({"name":"OVERLAY","inputs":[source,"x",2,1]}),
+ "EQUAL",
+ json!(["axcabc"]),
+ vec![1],
+ ),
+ (
+ json!({"name":"NULL"}),
+ "IS_NULL",
+ json!([]),
+ vec![0, 1, 2, 3],
+ ),
+ ];
+ for (transform, function, literals, expected) in cases {
+ let rule = leaf(transform, function, literals);
+ assert_eq!(kept(rule.clone()), expected, "{rule}");
+ }
+ }
+
+ #[test]
+ fn transformed_leaf_functions_keep_java_null_semantics() {
+ let transform =
json!({"name":"LENGTH","inputs":[input("s","STRING")]});
+ for (function, literals, expected) in [
+ ("EQUAL", json!([3]), vec![0]),
+ ("NOT_EQUAL", json!([3]), vec![1, 2]),
+ ("LESS_THAN", json!([3]), vec![2]),
+ ("LESS_OR_EQUAL", json!([3]), vec![0, 2]),
+ ("GREATER_THAN", json!([3]), vec![1]),
+ ("GREATER_OR_EQUAL", json!([3]), vec![0, 1]),
+ ("IN", json!([3, null]), vec![0]),
+ ("NOT_IN", json!([3]), vec![1, 2]),
+ ("NOT_IN", json!([3, null]), vec![]),
+ ("NOT_BETWEEN", json!([1, 4]), vec![1, 2]),
+ ("IS_NULL", json!([]), vec![3]),
+ ("IS_NOT_NULL", json!([]), vec![0, 1, 2]),
+ ("TRUE", json!([]), vec![0, 1, 2, 3]),
+ ("FALSE", json!([]), vec![]),
+ ("EQUAL", json!([null]), vec![]),
+ ("NOT_EQUAL", json!([null]), vec![]),
+ ] {
+ assert_eq!(
+ kept(leaf(transform.clone(), function, literals)),
+ expected,
+ "{function}"
+ );
+ }
+ let upper = json!({"name":"UPPER","inputs":[input("s","STRING")]});
+ for (function, literals) in [
+ ("STARTS_WITH", json!(["AB"])),
+ ("ENDS_WITH", json!(["BC"])),
+ ("CONTAINS", json!(["BCA"])),
+ ("LIKE", json!(["A_C%"])),
+ ] {
+ assert_eq!(
+ kept(leaf(upper.clone(), function, literals)),
+ vec![1],
+ "{function}"
+ );
+ }
+ }
+
+ #[test]
+ fn casts_compare_java_character_and_byte_widths() {
+ for (type_, literal, expected) in [
+ ("VARCHAR(1)", json!("η²"), vec![0]),
+ ("CHAR(8)", json!("abcabc "), vec![1]),
+ ("VARCHAR", json!("a"), vec![1]),
+ ("CHAR", json!(" "), vec![2]),
+ ] {
+ let transform =
json!({"name":"CAST","fieldRef":input("s","STRING"),"type":type_});
+ assert_eq!(
+ kept(leaf(transform, "EQUAL", json!([literal]))),
+ expected,
+ "{type_}"
+ );
+ }
+ for type_ in ["BINARY(8)", "VARBINARY(1)"] {
+ let transform =
json!({"name":"CAST","fieldRef":input("s","STRING"),"type":type_});
+ assert_eq!(kept(leaf(transform.clone(), "IS_NULL", json!([]))),
vec![3]);
+ assert_eq!(
+ kept(leaf(transform.clone(), "EQUAL", json!([null]))),
+ Vec::<i32>::new()
+ );
+ // Java does not deserialize non-NULL binary predicate literals,
+ // even when the function would ignore them.
+ for function in ["EQUAL", "TRUE", "IS_NULL"] {
+ let rule = leaf(transform.clone(), function, json!(["YQAA"]));
+ assert!(RowFilter::parse(&rule.to_string(),
&fields()).is_err());
+ }
+ }
+ }
+
+ #[test]
+ fn calendar_outputs_are_ints_and_keep_all_source_dependencies() {
+ let rule = leaf(
+ json!({"name":"YEAR_OF_WEEK","fieldRef":input("d","DATE")}),
+ "EQUAL",
+ json!([2025]),
+ );
+ assert_eq!(kept(rule.clone()), vec![1]);
+ let filter = RowFilter::parse(&rule.to_string(), &fields()).unwrap();
+ let mut deps = HashSet::new();
+ filter.collect_field_indices(&mut deps);
+ assert_eq!(deps, HashSet::from([3]));
+ let compound = json!({"kind":"COMPOUND","function":"AND","children":[
+ rule,
leaf(json!({"name":"SUBSTRING","inputs":[input("s","STRING"),input("p","BIGINT"),1]}),"EQUAL",json!(["a"]))
+ ]});
+ let mut deps = HashSet::new();
+ RowFilter::parse(&compound.to_string(), &fields())
+ .unwrap()
+ .collect_field_indices(&mut deps);
+ assert_eq!(deps, HashSet::from([1, 2, 3]));
+ }
+
+ #[test]
+ fn and_or_and_filter_list_short_circuit_before_throwing_transforms() {
+ let fields = fields();
+ let base = rows();
+ let mut columns = base.columns().to_vec();
+ columns[2] = Arc::new(Int64Array::from(vec![i64::MAX, 1, 0,
i64::MAX]));
+ let source = RecordBatch::try_new(base.schema(), columns).unwrap();
+ let guard = |id| {
+ leaf(
+ json!({"name":"FIELD_REF","fieldRef":input("id","INT")}),
+ "EQUAL",
+ json!([id]),
+ )
+ };
+ let substring = leaf(
+
json!({"name":"SUBSTRING","inputs":[input("s","STRING"),input("p","BIGINT"),1]}),
+ "EQUAL",
+ json!(["a"]),
+ );
+ for (function, id, expected) in [("AND", 1, vec![1]), ("OR", 0,
vec![0, 1])] {
+ let rule =
+
json!({"kind":"COMPOUND","function":function,"children":[guard(id),substring]});
+ let filter = RowFilter::parse(&rule.to_string(), &fields).unwrap();
+ let result = filter_batch(&source, &[filter], &fields,
&fields).unwrap();
+ assert_eq!(
+ result
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec(),
+ expected
+ );
+ }
+ let nested = json!({"kind":"COMPOUND","function":"OR","children":[
+ guard(0),
{"kind":"COMPOUND","function":"AND","children":[guard(1),substring]}
+ ]});
+ let nested = RowFilter::parse(&nested.to_string(), &fields).unwrap();
+ let result = filter_batch(&source, &[nested], &fields,
&fields).unwrap();
+ assert_eq!(
+ result
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec(),
+ vec![0, 1]
+ );
+ let filters = [guard(1), substring.clone()]
+ .map(|rule| RowFilter::parse(&rule.to_string(), &fields).unwrap());
+ assert_eq!(
+ filter_batch(&source, &filters, &fields, &fields)
+ .unwrap()
+ .num_rows(),
+ 1
+ );
+ assert!(filter_batch(
+ &source,
+ &[RowFilter::parse(&substring.to_string(), &fields).unwrap()],
+ &fields,
+ &fields
+ )
+ .is_err());
+ // Even a constant/NULL-literal function must execute its transform
first.
+ for (function, literals) in [
+ ("TRUE", json!([])),
+ ("FALSE", json!([])),
+ ("EQUAL", json!([null])),
+ ("NOT_EQUAL", json!([null])),
+ ] {
+ let rule = leaf(substring["transform"].clone(), function,
literals);
+ let filter = RowFilter::parse(&rule.to_string(), &fields).unwrap();
+ assert!(
+ filter_batch(&source, &[filter], &fields, &fields).is_err(),
+ "{function}"
+ );
+ }
+ }
+
+ #[test]
+ fn all_literals_and_unreachable_definitions_are_validated() {
+ let fields = fields();
+ for function in [
+ "TRUE",
+ "FALSE",
+ "IS_NULL",
+ "IS_NOT_NULL",
+ "EQUAL",
+ "NOT_EQUAL",
+ "IN",
+ ] {
+ for literals in [json!(["bad"]), json!([null, "bad"])] {
+ let rule = leaf(
+ json!({"name":"LENGTH","inputs":[input("s","STRING")]}),
+ function,
+ literals,
+ );
+ assert!(
+ RowFilter::parse(&rule.to_string(), &fields).is_err(),
+ "{rule}"
+ );
+ let unreachable =
json!({"kind":"COMPOUND","function":"OR","children":[
+ leaf(json!({"name":"NULL"}),"TRUE",json!([])),rule
+ ]});
+ assert!(RowFilter::parse(&unreachable.to_string(),
&fields).is_err());
+ }
+ }
+ for function in ["TRUE", "FALSE", "IS_NULL", "IS_NOT_NULL"] {
+ let mut rule = leaf(
+ json!({"name":"LENGTH","inputs":[input("s","STRING")]}),
+ function,
+ Value::Null,
+ );
+ assert!(RowFilter::parse(&rule.to_string(), &fields).is_ok());
+ rule.as_object_mut().unwrap().remove("literals");
+ assert!(RowFilter::parse(&rule.to_string(), &fields).is_ok());
+ }
+ let rebound = leaf(
+
json!({"name":"CAST","fieldRef":input("s","BOOLEAN"),"type":"STRING"}),
+ "TRUE",
+ json!([]),
+ );
+ assert!(RowFilter::parse(&rebound.to_string(), &fields)
+ .unwrap_err()
+ .to_string()
+ .contains("rebound source"));
+ }
+}
diff --git a/crates/paimon/src/table/query_auth/rules.rs
b/crates/paimon/src/table/query_auth/rules.rs
index 5d5ccd4f..004e8b76 100644
--- a/crates/paimon/src/table/query_auth/rules.rs
+++ b/crates/paimon/src/table/query_auth/rules.rs
@@ -18,6 +18,7 @@
//! Apply the server's row filters and column masks to read batches.
use super::unsupported;
+use super::RowFilter;
use crate::api::AuthTableQueryResponse;
use crate::arrow::residual::{evaluate_exact_leaf_predicate,
sanitize_filter_mask};
use crate::spec::{DataField, Predicate, Transform, TransformInput};
@@ -37,7 +38,7 @@ pub(crate) struct ColumnMask {
/// One grant's rules, indexed by table schema.
#[derive(Debug, Default, PartialEq)]
pub(crate) struct Rules {
- pub(crate) filters: Vec<Predicate>,
+ pub(crate) filters: Vec<RowFilter>,
pub(crate) masks: Vec<ColumnMask>,
}
@@ -50,15 +51,15 @@ impl Rules {
return Err(unsupported("the server sent an empty row filter"));
}
let filter =
- Predicate::from_rest_json(json, fields).map_err(
- |e| match mentioned_system_column(json) {
+ RowFilter::parse(json, fields).map_err(|e| {
+ match mentioned_system_column(json) {
Some(name) => unsupported(&format!(
"the server's row filter reads the system column
'{name}', which a \
query-auth read never projects"
)),
None => unsupported(&format!("cannot parse the
server's row filter: {e}")),
- },
- )?;
+ }
+ })?;
filters.push(filter);
}
@@ -130,7 +131,7 @@ impl Rules {
pub(crate) fn filter_columns(&self) -> HashSet<usize> {
let mut out = HashSet::new();
for filter in &self.filters {
- filter.collect_leaf_field_indices(&mut out);
+ filter.collect_field_indices(&mut out);
}
out
}
@@ -160,13 +161,16 @@ fn check_mask_fits(transform: &Transform, target:
&DataField, fields: &[DataFiel
target.name()
)));
}
+ check_transform_cast(transform, fields)
+}
+
+pub(super) fn check_transform_cast(transform: &Transform, fields:
&[DataField]) -> Result<()> {
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()
+ "the server's transform casts {from:?} to {to:?}, which this
client does not \
+ convert as Java does"
)));
}
}
@@ -216,28 +220,28 @@ fn mentioned_system_column(json: &str) -> Option<&'static
str> {
/// cannot be evaluated is an error, never a kept row.
pub(crate) fn filter_batch(
batch: &RecordBatch,
- filters: &[Predicate],
+ filters: &[RowFilter],
schema_fields: &[DataField],
batch_fields: &[DataField],
) -> Result<RecordBatch> {
- let mut keep: Option<BooleanArray> = None;
+ let mut result = batch.clone();
+ // Java ANDs the server's entries in order, so later transforms never see
+ // a row an earlier rule rejected.
for filter in filters {
- let mask = rule_mask(batch, filter, schema_fields, batch_fields)?;
- keep = Some(match keep {
- Some(existing) => combine(&existing, &mask, false)?,
- None => mask,
- });
+ if result.num_rows() == 0 {
+ break;
+ }
+ let keep = filter.evaluate(&result, schema_fields, batch_fields)?;
+ result =
+ arrow_select::filter::filter_record_batch(&result,
&keep).map_err(|e| eval_err(&e))?;
}
- let Some(keep) = keep else {
- return Ok(batch.clone());
- };
- arrow_select::filter::filter_record_batch(batch, &keep).map_err(|e|
eval_err(&e))
+ Ok(result)
}
/// Java leaf semantics: a null value fails every leaf but `IS NULL`, for the
/// negated functions too. The REST parser builds no `NOT`, so two-valued
/// AND/OR over those leaves is exactly Java's `test`.
-fn rule_mask(
+pub(super) fn rule_mask(
batch: &RecordBatch,
predicate: &Predicate,
schema_fields: &[DataField],
@@ -288,7 +292,7 @@ fn combine(left: &BooleanArray, right: &BooleanArray, or:
bool) -> Result<Boolea
}
/// The column holding `field`, matched by id and name so an alias cannot
stand in.
-fn column_of(
+pub(super) fn column_of(
batch: &RecordBatch,
field: &DataField,
batch_fields: &[DataField],
@@ -312,7 +316,7 @@ fn column_of(
/// Java's `Float.compare` puts every NaN above every number; Arrow's total
/// order puts a negative NaN below them, so `f < 0` would admit it.
-fn canonicalize_nan(column: &ArrayRef) -> ArrayRef {
+pub(super) fn canonicalize_nan(column: &ArrayRef) -> ArrayRef {
match column.data_type() {
arrow_schema::DataType::Float32 => match
column.as_any().downcast_ref::<Float32Array>() {
Some(values) if values.iter().any(|v| v.is_some_and(f32::is_nan))
=> {
@@ -379,25 +383,7 @@ pub(crate) fn mask_batch(
)));
};
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::DateExtract(index, part) => {
- crate::arrow::evaluate_date_extract(&input(*index)?, *part)?
- }
- Transform::Upper(_)
- | Transform::Lower(_)
- | Transform::Concat(_)
- | Transform::ConcatWs(_) => string_mask(batch, &mask.transform,
&input)?,
- _ => {
- crate::arrow::evaluate_string_transform(&mask.transform,
batch.num_rows(), &input)?
- }
- };
+ let masked = evaluate_transform(&mask.transform, batch.num_rows(),
&input, &target_type)?;
let masked = cast_to(&masked, &target_type)?;
for position in targets {
columns[position] = Arc::clone(&masked);
@@ -411,6 +397,53 @@ pub(crate) fn mask_batch(
.map_err(|e| eval_err(&e))
}
+/// Execute one expression on complete input fields. Both filters and masks use
+/// this path, so width, NULL and calendar semantics cannot drift between them.
+pub(super) fn evaluate_transform(
+ transform: &Transform,
+ rows: usize,
+ input: &dyn Fn(usize) -> Result<ArrayRef>,
+ null_type: &arrow_schema::DataType,
+) -> Result<ArrayRef> {
+ match transform {
+ Transform::Null => Ok(arrow_array::new_null_array(null_type, rows)),
+ Transform::FieldRef(index) => input(*index),
+ Transform::Cast(index, to) => {
+ let cast = cast_to(&input(*index)?,
&crate::arrow::paimon_type_to_arrow(to)?)?;
+ Ok(apply_declared_width(&cast, to))
+ }
+ Transform::DateExtract(index, part) => {
+ crate::arrow::evaluate_date_extract(&input(*index)?, *part)
+ }
+ Transform::Upper(_)
+ | Transform::Lower(_)
+ | Transform::Concat(_)
+ | Transform::ConcatWs(_) => string_mask(rows, transform, input),
+ _ => crate::arrow::evaluate_string_transform(transform, rows, input),
+ }
+}
+
+pub(super) fn transform_output_type(
+ transform: &Transform,
+ fields: &[DataField],
+) -> Result<crate::spec::DataType> {
+ use crate::spec::{BooleanType, DataType, IntType, VarCharType};
+ Ok(match transform {
+ // NullTransform.outputType is BOOLEAN, except a mask adopts its
target.
+ Transform::Null => DataType::Boolean(BooleanType::new()),
+ Transform::FieldRef(index) => fields
+ .get(*index)
+ .ok_or_else(|| unsupported("the server's transform reads an
unknown column"))?
+ .data_type()
+ .clone(),
+ Transform::Cast(_, to) => to.clone(),
+ Transform::Length(_) | Transform::BitLength(_) |
Transform::DateExtract(_, _) => {
+ DataType::Int(IntType::new())
+ }
+ _ => DataType::VarChar(VarCharType::string_type()),
+ })
+}
+
fn mask_can_be_null(transform: &Transform, fields: &[DataField]) -> bool {
let nullable = |input: &TransformInput| match input {
TransformInput::Literal(literal) => literal.is_none(),
@@ -431,16 +464,12 @@ fn mask_can_be_null(transform: &Transform, fields:
&[DataField]) -> bool {
/// `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::Length(_) | Transform::BitLength(_) |
Transform::DateExtract(_, _) => {
- Some(arrow_schema::DataType::Int32)
- }
- _ => Some(arrow_schema::DataType::Utf8),
+ if matches!(transform, Transform::Null) {
+ None
+ } else {
+ transform_output_type(transform, fields)
+ .and_then(|t| crate::arrow::paimon_type_to_arrow(&t))
+ .ok()
}
}
@@ -525,7 +554,7 @@ impl StringInput<'_> {
}
fn string_mask(
- batch: &RecordBatch,
+ rows: usize,
transform: &Transform,
input: &dyn Fn(usize) -> Result<ArrayRef>,
) -> Result<ArrayRef> {
@@ -558,7 +587,7 @@ fn string_mask(
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);
+ let mut values = StringBuilder::with_capacity(rows, 0);
match literal {
Some(value) => values.append_value_n(
if upper {
@@ -566,14 +595,14 @@ fn string_mask(
} else {
value.to_lowercase()
},
- batch.num_rows(),
+ rows,
),
- None => values.append_nulls(batch.num_rows()),
+ None => values.append_nulls(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 mut values = StringBuilder::with_capacity(rows, 0);
+ for row in 0..rows {
let (parts, separator) = if matches!(transform,
Transform::ConcatWs(_)) {
let Some(separator) = resolved[0].value(row) else {
values.append_null();
@@ -849,6 +878,15 @@ mod tests {
]
}
+ #[test]
+ fn transformed_row_filter_reads_hidden_source_as_java_does() {
+ let fields = typed_fields();
+ let filter =
r#"{"kind":"LEAF","transform":{"name":"UPPER","inputs":[{"name":"s","type":"STRING","index":99}]},"function":"EQUAL","literals":["BANANA"]}"#;
+ let rules = parse(&[filter], &[], &fields).unwrap();
+ assert_eq!(rules.filter_columns(), HashSet::from([5]));
+ assert_eq!(kept(&rules.filters), vec![2]);
+ }
+
#[test]
fn test_parse_reads_the_rules() {
let fields = rule_fields();
@@ -1083,7 +1121,7 @@ mod tests {
}
/// The rows of the typed batch that `filters` keep.
- fn kept(filters: &[Predicate]) -> Vec<i32> {
+ fn kept(filters: &[RowFilter]) -> Vec<i32> {
let fields = typed_fields();
ids(&filter_batch(&typed_batch(), filters, &fields, &fields).unwrap())
}
@@ -1186,11 +1224,11 @@ mod tests {
(b.is_not_in("ts", vec![ts(1000, 0)]), vec![1, 2, 3]),
];
for (predicate, expected) in cases {
- let predicate = predicate.unwrap();
+ let predicate = RowFilter::Predicate(predicate.unwrap());
assert_eq!(
kept(std::slice::from_ref(&predicate)),
expected,
- "{predicate}"
+ "{predicate:?}"
);
}
}
@@ -1320,11 +1358,16 @@ mod tests {
literals: vec![Datum::String("1".to_string())],
};
for (filter, reason) in [
- (Predicate::Not(Box::new(filters[0].clone())), "has a NOT"),
+ (Predicate::Not(Box::new(Predicate::AlwaysTrue)), "has a NOT"),
(Predicate::And(Vec::new()), "empty AND/OR"),
(mistyped, "cannot evaluate"),
] {
- let message = refusal(filter_batch(&rows, &[filter], &fields,
&fields));
+ let message = refusal(filter_batch(
+ &rows,
+ &[RowFilter::Predicate(filter)],
+ &fields,
+ &fields,
+ ));
assert!(message.contains(reason), "{message}");
}
let message = refusal(filter_batch(&rows, &filters, &fields[..1],
&fields));
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index 593f432c..623e1419 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -1812,7 +1812,7 @@ mod tests {
};
let s = || split(vec![file("a", 10, None)], true);
assert_eq!(s().planned(grant(Vec::new())).merged_row_count(),
Some(10));
- let filtered =
s().planned(grant(vec![crate::spec::Predicate::AlwaysTrue]));
+ let filtered =
s().planned(grant(vec![crate::spec::Predicate::AlwaysTrue.into()]));
assert_eq!(filtered.merged_row_count(), None);
// The physical count stays, as Java's `rowCount` does.
assert_eq!(filtered.row_count(), 10);
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 791336d6..74fa261f 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -2897,7 +2897,7 @@ mod tests {
None,
crate::table::query_auth::Rules {
filters: if restricted {
- vec![crate::spec::Predicate::AlwaysTrue]
+ vec![crate::spec::Predicate::AlwaysTrue.into()]
} else {
Vec::new()
},
@@ -3160,7 +3160,10 @@ mod tests {
table.query_auth_session().unwrap(),
Some(Vec::new()),
crate::table::query_auth::Rules {
- filters,
+ filters: filters
+ .into_iter()
+ .map(crate::table::query_auth::RowFilter::from)
+ .collect(),
masks: Vec::new(),
},
);
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 356e5c32..323c05d6 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -4519,6 +4519,86 @@ async fn
test_query_auth_calendar_masks_read_hidden_operands_before_filter_and_l
}
}
+#[cfg(not(windows))]
+#[tokio::test]
+async fn
test_query_auth_transformed_filter_precedes_mask_user_filter_and_limit() {
+ use arrow_array::Date32Array;
+ use paimon::spec::DateType;
+ for (primary_key, evolution) in [(false, false), (true, false), (false,
true)] {
+ let schema = |options: &[(&str, &str)]| {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::VarChar(VarCharType::string_type()))
+ .column("source",
DataType::VarChar(VarCharType::string_type()))
+ .column("p", DataType::BigInt(BigIntType::new()))
+ .column("d", DataType::Date(DateType::new()));
+ 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");
+ }
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().unwrap()
+ };
+ let source = RecordBatch::try_from_iter(vec![
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4])) as Arc<dyn Array>,
+ ),
+ ("value", Arc::new(StringArray::from(vec!["raw"; 4]))),
+ (
+ "source",
+ Arc::new(StringArray::from(vec![
+ Some("old"),
+ Some("abc"),
+ Some("xyz"),
+ None,
+ ])),
+ ),
+ (
+ "p",
+ Arc::new(Int64Array::from(vec![i64::MAX, 1, i64::MAX,
i64::MAX])),
+ ),
+ (
+ "d",
+ Arc::new(Date32Array::from(vec![
+ Some(19782),
+ Some(20087),
+ Some(-1),
+ None,
+ ])),
+ ),
+ ])
+ .unwrap();
+ let g = written("transformed_filter", schema, vec![source]).await;
+ let filter =
serde_json::json!({"kind":"COMPOUND","function":"AND","children":[
+
{"kind":"LEAF","transform":{"name":"YEAR_OF_WEEK","fieldRef":{"name":"d","type":"DATE","index":99}},"function":"EQUAL","literals":[2025]},
+
{"kind":"LEAF","transform":{"name":"SUBSTRING","inputs":[{"name":"source","type":"STRING"},{"name":"p","type":"BIGINT"},1]},"function":"EQUAL","literals":["a"]}
+ ]});
+ g.set_rules(
+ &[filter],
+ &[(
+ "value",
+ serde_json::json!({"name":"CONCAT","inputs":["visible"]}),
+ )],
+ );
+ let mut builder = g.table.new_read_builder();
+ builder.with_projection(&["id"]).unwrap();
+ builder.with_filter(
+ PredicateBuilder::new(g.table.schema().fields())
+ .equal("value", Datum::String("visible".into()))
+ .unwrap(),
+ );
+ builder.with_limit(1);
+ assert_eq!(ids(&read_all(&builder).await.unwrap()), vec![2]);
+ }
+}
+
#[cfg(not(windows))]
#[tokio::test]
async fn test_query_auth_widens_an_integer_mask() {
