diff --git a/crates/paimon/src/spec/types.rs b/crates/paimon/src/spec/types.rs index ab73da7e3..0ff006639 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 3de5625a5..746661291 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 000000000..7c8afec65 --- /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), + Or(Vec), +} + +impl RowFilter { + pub(super) fn parse(json: &str, fields: &[DataField]) -> Result { + 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 { + 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::>>()?; + 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) { + 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 { + match self { + Self::Predicate(predicate) => rule_mask(batch, predicate, fields, batch_fields), + Self::Transformed { + transform, + predicate, + data_type, + } => { + let input = |index: usize| -> Result { + 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 = (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::>()); + 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 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 { + 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 { + 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::() + .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::::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::() + .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::() + .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 5d5ccd4f3..004e8b76f 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, + pub(crate) filters: Vec, pub(crate) masks: Vec, } @@ -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 { 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 { - let mut keep: Option = 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 ArrayRef { +pub(super) fn canonicalize_nan(column: &ArrayRef) -> ArrayRef { match column.data_type() { arrow_schema::DataType::Float32 => match column.as_any().downcast_ref::() { 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, + null_type: &arrow_schema::DataType, +) -> Result { + 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 { + 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 { - 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, ) -> Result { @@ -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 { + fn kept(filters: &[RowFilter]) -> Vec { 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 593f432c1..623e1419e 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 791336d62..74fa261fe 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 356e5c320..323c05d66 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, + ), + ("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() {