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 31201c07 [read] Support Java-compatible REST query authorization masks
(#1122)
31201c07 is described below
commit 31201c0717bd7a9b9753174b8a473f73186ba04c
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 13 20:10:25 2026 +0800
[read] Support Java-compatible REST query authorization masks (#1122)
---
crates/paimon/src/arrow/mod.rs | 2 +
crates/paimon/src/arrow/string_transform.rs | 447 +++++++++++++++++++++
crates/paimon/src/spec/mod.rs | 4 +-
crates/paimon/src/spec/predicate.rs | 225 ++++++++++-
crates/paimon/src/spec/schema.rs | 33 +-
.../src/table/goldens/split_v1_query_auth.bin | Bin 0 -> 993 bytes
crates/paimon/src/table/mod.rs | 22 +
crates/paimon/src/table/query_auth.rs | 95 ++++-
crates/paimon/src/table/query_auth/rules.rs | 62 ++-
crates/paimon/src/table/source.rs | 268 +++++++++++-
10 files changed, 1113 insertions(+), 45 deletions(-)
diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs
index 34140b97..7273e175 100644
--- a/crates/paimon/src/arrow/mod.rs
+++ b/crates/paimon/src/arrow/mod.rs
@@ -26,6 +26,8 @@ pub(crate) mod partition;
mod read_budget;
pub(crate) mod residual;
mod row_filter;
+mod string_transform;
+pub(crate) use string_transform::evaluate_string_transform;
pub(crate) mod schema_evolution;
pub(crate) mod shredding;
mod variant;
diff --git a/crates/paimon/src/arrow/string_transform.rs
b/crates/paimon/src/arrow/string_transform.rs
new file mode 100644
index 00000000..71d5dbab
--- /dev/null
+++ b/crates/paimon/src/arrow/string_transform.rs
@@ -0,0 +1,447 @@
+// 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.
+
+//! Arrow execution of Java's character and position transforms.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use crate::spec::{PadDirection, Transform, TransformInput, TrimFlag};
+use crate::Result;
+use arrow_array::{cast::AsArray, Array, ArrayRef, Int32Array, Int64Array,
StringArray};
+use arrow_schema::DataType as ArrowType;
+
+enum Input<'a> {
+ Text(Option<&'a str>),
+ Number(Option<i64>),
+ IntegerText(&'a str),
+ Strings(StringArray),
+ Integers(Int64Array),
+}
+
+impl Input<'_> {
+ fn is_null(&self, row: usize) -> bool {
+ match self {
+ Self::Text(value) => value.is_none(),
+ Self::Number(value) => value.is_none(),
+ Self::IntegerText(_) => false,
+ Self::Strings(values) => values.is_null(row),
+ Self::Integers(values) => values.is_null(row),
+ }
+ }
+ fn text(&self, row: usize) -> Result<Option<&str>> {
+ match self {
+ Self::Text(value) => Ok(*value),
+ Self::Strings(values) => Ok(values.is_valid(row).then(||
values.value(row))),
+ _ => Err(invalid("string transform requires a character input")),
+ }
+ }
+
+ fn integer(&self, row: usize) -> Result<Option<i32>> {
+ let value = match self {
+ Self::IntegerText(text) => return parse_integer(text).map(Some),
+ Self::Number(value) => *value,
+ Self::Integers(values) => values.is_valid(row).then(||
values.value(row)),
+ _ => return Err(invalid("position transform requires an integer
input")),
+ };
+ value
+ .map(|v| {
+ i32::try_from(v).map_err(|_| invalid("transform position
exceeds Java INT range"))
+ })
+ .transpose()
+ }
+}
+
+pub(crate) fn evaluate_string_transform(
+ transform: &Transform,
+ rows: usize,
+ column: &dyn Fn(usize) -> Result<ArrayRef>,
+) -> Result<ArrayRef> {
+ let inputs = transform
+ .inputs()
+ .iter()
+ .map(|input| match input {
+ TransformInput::Literal(value) =>
Ok(Input::Text(value.as_deref())),
+ TransformInput::Integer(value) => Ok(Input::Number(*value)),
+ TransformInput::IntegerText(text) => Ok(Input::IntegerText(text)),
+ TransformInput::Field(index) => {
+ let array = column(*index)?;
+ match array.data_type() {
+ ArrowType::Utf8 =>
Ok(Input::Strings(array.as_string::<i32>().clone())),
+ ArrowType::Int8 | ArrowType::Int16 | ArrowType::Int32 |
ArrowType::Int64 => {
+ let array = arrow_cast::cast(&array, &ArrowType::Int64)
+ .map_err(|e| invalid(&e.to_string()))?;
+ Ok(Input::Integers(
+ array
+
.as_primitive::<arrow_array::types::Int64Type>()
+ .clone(),
+ ))
+ }
+ other => Err(invalid(&format!("invalid transform input
type {other:?}"))),
+ }
+ }
+ })
+ .collect::<Result<Vec<_>>>()?;
+ if matches!(transform, Transform::Length(_) | Transform::BitLength(_)) {
+ let values = (0..rows)
+ .map(|row| {
+ Ok(inputs[0].text(row)?.map(|text| {
+ if matches!(transform, Transform::Length(_)) {
+ text.chars().count() as i32
+ } else {
+ (text.len() as i32).wrapping_mul(8)
+ }
+ }))
+ })
+ .collect::<Result<Vec<_>>>()?;
+ return Ok(Arc::new(Int32Array::from(values)));
+ }
+ let values = (0..rows)
+ .map(|row| evaluate_row(transform, &inputs, row))
+ .collect::<Result<Vec<_>>>()?;
+ Ok(Arc::new(StringArray::from(values)))
+}
+
+fn evaluate_row(transform: &Transform, inputs: &[Input<'_>], row: usize) ->
Result<Option<String>> {
+ let source = inputs[0].text(row)?;
+ let value = match transform {
+ Transform::Substring(_) => {
+ // SubstringTransform checks source and null positions before
+ // evaluating positions; a hidden NULL source cannot raise
overflow.
+ let Some(source) = source else {
+ return Ok(None);
+ };
+ if inputs[1..].iter().any(|input| input.is_null(row)) {
+ return Ok(None);
+ }
+ let Some(position) = inputs[1].integer(row)? else {
+ return Ok(None);
+ };
+ let length = if inputs.len() == 3 {
+ let Some(length) = inputs[2].integer(row)? else {
+ return Ok(None);
+ };
+ length
+ } else {
+ i32::MAX
+ };
+ substring_sql(source, position, length).to_owned()
+ }
+ Transform::Trim(_, flag) => {
+ let Some(source) = source else {
+ return Ok(None);
+ };
+ let chars = if inputs.len() == 2 {
+ let Some(chars) = inputs[1].text(row)? else {
+ return Ok(None);
+ };
+ chars
+ } else {
+ " "
+ };
+ let contains = |c| chars.contains(c);
+ match flag {
+ TrimFlag::Leading => source.trim_start_matches(contains),
+ TrimFlag::Trailing => source.trim_end_matches(contains),
+ TrimFlag::Both => source.trim_matches(contains),
+ }
+ .to_owned()
+ }
+ Transform::Pad(_, direction) => {
+ let length = inputs[1].integer(row)?;
+ let padding = inputs[2].text(row)?;
+ let (Some(source), Some(length), Some(padding)) = (source, length,
padding) else {
+ return Ok(None);
+ };
+ let count = source.chars().count() as i32;
+ let needed = length.wrapping_sub(count);
+ if needed <= 0 || padding.is_empty() {
+ substring(source, 0, length).to_owned()
+ } else {
+ let pad_chars = padding.chars().count();
+ let repeats = needed as usize / pad_chars;
+ let tail = substring(padding, 0, needed % pad_chars as i32);
+ let size = padding
+ .len()
+ .checked_mul(repeats)
+ .and_then(|n| n.checked_add(tail.len()))
+ .and_then(|n| n.checked_add(source.len()))
+ .filter(|n| *n <= i32::MAX as usize)
+ .ok_or_else(|| invalid("PAD result exceeds Arrow UTF-8
storage"))?;
+ let mut result = String::new();
+ result
+ .try_reserve(size)
+ .map_err(|e| invalid(&format!("cannot reserve PAD result:
{e}")))?;
+ if *direction == PadDirection::Right {
+ result.push_str(source);
+ }
+ for _ in 0..repeats {
+ result.push_str(padding);
+ }
+ result.push_str(tail);
+ if *direction == PadDirection::Left {
+ result.push_str(source);
+ }
+ result
+ }
+ }
+ Transform::Translate(_) => {
+ let matching = inputs[1].text(row)?;
+ let replacement = inputs[2].text(row)?;
+ let (Some(source), Some(matching), Some(replacement)) = (source,
matching, replacement)
+ else {
+ return Ok(None);
+ };
+ let replacements: Vec<char> = replacement.chars().collect();
+ let mut mapping = HashMap::new();
+ for (index, ch) in matching.chars().enumerate() {
+ mapping
+ .entry(ch)
+ .or_insert(replacements.get(index).copied().filter(|c| *c
!= '\0'));
+ }
+ source
+ .chars()
+ .filter_map(|ch| mapping.get(&ch).copied().unwrap_or(Some(ch)))
+ .collect()
+ }
+ Transform::Overlay(_) => {
+ let replacement = inputs[1].text(row)?;
+ let position = inputs[2].integer(row)?;
+ let length = if inputs.len() == 4 {
+ inputs[3].integer(row)?
+ } else {
+ None
+ };
+ let (Some(source), Some(replacement), Some(position)) = (source,
replacement, position)
+ else {
+ return Ok(None);
+ };
+ if inputs.len() == 4 && length.is_none() {
+ return Ok(None);
+ }
+ let replaced = length
+ .filter(|length| *length >= 0)
+ .unwrap_or_else(|| replacement.chars().count() as i32);
+ format!(
+ "{}{}{}",
+ substring_sql(source, 1, position.wrapping_sub(1)),
+ replacement,
+ substring_sql(source, position.wrapping_add(replaced),
i32::MAX)
+ )
+ }
+ _ => return Err(invalid("unsupported character transform")),
+ };
+ Ok(Some(value))
+}
+
+fn substring_sql(source: &str, position: i32, length: i32) -> &str {
+ let count = source.chars().count() as i32;
+ let start = if position > 0 {
+ position - 1
+ } else if position < 0 {
+ count + position
+ } else {
+ 0
+ };
+ let end = (i64::from(start) +
i64::from(length)).clamp(i64::from(i32::MIN), i64::from(i32::MAX))
+ as i32;
+ substring(source, start, end)
+}
+
+fn substring(source: &str, start: i32, end: i32) -> &str {
+ if end <= start || end <= 0 {
+ return "";
+ }
+ let start = source
+ .char_indices()
+ .nth(start.max(0) as usize)
+ .map_or(source.len(), |(index, _)| index);
+ let end = source
+ .char_indices()
+ .nth(end as usize)
+ .map_or(source.len(), |(index, _)| index);
+ &source[start..end.max(start)]
+}
+
+fn invalid(message: &str) -> crate::Error {
+ crate::Error::DataInvalid {
+ message: message.to_owned(),
+ source: None,
+ }
+}
+
+/// Integer.parseInt uses Character.digit on UTF-16 chars: decimal digits in
+/// BMP scripts are accepted, supplementary digits and whitespace are rejected.
+fn parse_integer(text: &str) -> Result<i32> {
+ const ZEROES: &[u32] = &[
+ 0x0030, 0x0660, 0x06f0, 0x07c0, 0x0966, 0x09e6, 0x0a66, 0x0ae6,
0x0b66, 0x0be6, 0x0c66,
+ 0x0ce6, 0x0d66, 0x0de6, 0x0e50, 0x0ed0, 0x0f20, 0x1040, 0x1090,
0x17e0, 0x1810, 0x1946,
+ 0x19d0, 0x1a80, 0x1a90, 0x1b50, 0x1bb0, 0x1c40, 0x1c50, 0xa620,
0xa8d0, 0xa900, 0xa9d0,
+ 0xa9f0, 0xaa50, 0xabf0, 0xff10,
+ ];
+ let (negative, digits) = match text.strip_prefix('-') {
+ Some(digits) => (true, digits),
+ None => (false, text.strip_prefix('+').unwrap_or(text)),
+ };
+ if digits.is_empty() {
+ return Err(invalid("empty transform position"));
+ }
+ // Accumulate negatively, so i32::MIN remains representable.
+ let mut value = 0i32;
+ for ch in digits.chars() {
+ let digit = ZEROES
+ .iter()
+ .find_map(|zero| {
+ (u32::from(ch) >= *zero && u32::from(ch) < zero + 10)
+ .then(|| (u32::from(ch) - zero) as i32)
+ })
+ .ok_or_else(|| invalid("invalid Java integer position"))?;
+ value = value
+ .checked_mul(10)
+ .and_then(|v| v.checked_sub(digit))
+ .ok_or_else(|| invalid("transform position exceeds Java INT
range"))?;
+ }
+ if negative {
+ Ok(value)
+ } else {
+ value
+ .checked_neg()
+ .ok_or_else(|| invalid("transform position exceeds Java INT
range"))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::{BigIntType, DataField, DataType, VarCharType};
+
+ fn apply(rule: &str, source: Vec<Option<&str>>) -> Result<ArrayRef> {
+ let fields = vec![DataField::new(
+ 0,
+ "s".into(),
+ DataType::VarChar(VarCharType::string_type()),
+ )];
+ let transform = Transform::from_rest_json(rule, &fields)?;
+ let rows = source.len();
+ let source = Arc::new(StringArray::from(source)) as ArrayRef;
+ evaluate_string_transform(&transform, rows, &|_| Ok(source.clone()))
+ }
+
+ #[test]
+ fn substring_positions_are_parsed_after_null_checks() {
+ for position in [
+ r#""bad""#,
+ r#""999999999999999999999""#,
+ "18446744073709551617",
+ ] {
+ let rule =
format!(r#"{{"name":"SUBSTRING","inputs":[{{"name":"s"}}, {position}]}}"#);
+ assert!(apply(&rule, vec![None]).unwrap().is_null(0));
+ assert!(apply(&rule, vec![Some("abc")]).is_err());
+ let rule =
+ format!(r#"{{"name":"SUBSTRING","inputs":[{{"name":"s"}},
{position}, null]}}"#);
+ assert!(apply(&rule, vec![Some("abc")]).unwrap().is_null(0));
+ }
+ let result = apply(
+ r#"{"name":"SUBSTRING","inputs":[{"name":"s"},"+٢",1]}"#,
+ vec![Some("abc")],
+ )
+ .unwrap();
+ assert_eq!(result.as_string::<i32>().value(0), "b");
+ for position in ["1.5", "1e2", "true"] {
+ assert!(apply(
+
&format!(r#"{{"name":"SUBSTRING","inputs":[null,{position}]}}"#),
+ vec![None]
+ )
+ .is_err());
+ }
+ }
+
+ #[test]
+ fn integer_syntax_and_range_match_java() {
+ for (text, value) in [
+ ("-2147483648", i32::MIN),
+ ("+2147483647", i32::MAX),
+ ("٠٠٢", 2),
+ ("２", 2),
+ ] {
+ assert_eq!(parse_integer(text).unwrap(), value);
+ }
+ for text in ["", "+", " 2", "2 ", "1_0", "2147483648", "-2147483649",
"𝟚"] {
+ assert!(parse_integer(text).is_err(), "{text}");
+ }
+ }
+
+ #[test]
+ fn pad_and_overlay_convert_integers_before_null_strings() {
+ for rule in [
+
r#"{"name":"PAD","direction":"LEFT","inputs":[null,2147483648,"x"]}"#,
+ r#"{"name":"OVERLAY","inputs":[null,"x",2147483648,null]}"#,
+ r#"{"name":"OVERLAY","inputs":[null,"x",null,2147483648]}"#,
+ ] {
+ assert!(apply(rule, vec![None]).is_err());
+ }
+ }
+
+ #[test]
+ fn mixed_fields_are_bound_by_name_and_checked_before_execution() {
+ let fields = vec![
+ DataField::new(0, "s".into(),
DataType::VarChar(VarCharType::string_type())),
+ DataField::new(1, "n".into(), DataType::BigInt(BigIntType::new())),
+ ];
+ for rule in [
+
r#"{"name":"PAD","direction":"RIGHT","inputs":[{"name":"s","type":"INT"},3,"x"]}"#,
+
r#"{"name":"PAD","direction":"RIGHT","inputs":["a",{"name":"n","type":"STRING"},"x"]}"#,
+
r#"{"name":"OVERLAY","inputs":["a","x",{"name":"n","type":"STRING"}]}"#,
+
r#"{"name":"TRIM","trimFlag":"BOTH","inputs":[{"name":"s","type":"INT"}]}"#,
+ r#"{"name":"SUBSTRING","inputs":["a",{"name":"s","type":"INT"}]}"#,
+ ] {
+ assert!(Transform::from_rest_json(rule, &fields).is_err(),
"{rule}");
+ }
+ let transform =
Transform::from_rest_json(r#"{"name":"SUBSTRING","inputs":[{"index":99,"name":"s","type":"INT"},{"index":99,"name":"n","type":"STRING"},1]}"#,
&fields).unwrap();
+ let columns: Vec<ArrayRef> = vec![
+ Arc::new(StringArray::from(vec![Some("甲🙂乙"), None])),
+ Arc::new(Int64Array::from(vec![2, i64::MAX])),
+ ];
+ let result =
+ evaluate_string_transform(&transform, 2, &|index|
Ok(columns[index].clone())).unwrap();
+ assert_eq!(
+ result.as_string::<i32>().iter().collect::<Vec<_>>(),
+ vec![Some("🙂"), None]
+ );
+ }
+
+ #[test]
+ fn lengths_count_unicode_characters_and_utf8_bits_as_int() {
+ for (name, expected) in [
+ ("LENGTH", vec![Some(3), Some(2), Some(0), None]),
+ ("BIT_LENGTH", vec![Some(80), Some(24), Some(0), None]),
+ ] {
+ let result = apply(
+
&format!(r#"{{"name":"{name}","inputs":[{{"name":"s","type":"STRING"}}]}}"#),
+ vec![Some("甲🙂乙"), Some("e\u{301}"), Some(""), None],
+ )
+ .unwrap();
+ assert_eq!(
+ result
+ .as_primitive::<arrow_array::types::Int32Type>()
+ .iter()
+ .collect::<Vec<_>>(),
+ expected
+ );
+ }
+ }
+}
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index 7969f9b3..a108f89e 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -133,8 +133,8 @@ pub(crate) use predicate::like_match;
pub(crate) use predicate::row_id_leaf;
pub(crate) use predicate::validate_datum_matches_type;
pub use predicate::{
- field_idx_to_partition_idx, Datum, Predicate, PredicateBuilder,
PredicateOperator, Transform,
- TransformInput,
+ field_idx_to_partition_idx, Datum, PadDirection, Predicate,
PredicateBuilder,
+ PredicateOperator, Transform, TransformInput, TrimFlag,
};
pub(crate) mod murmur_hash;
mod partition_statistics;
diff --git a/crates/paimon/src/spec/predicate.rs
b/crates/paimon/src/spec/predicate.rs
index 21c5ec81..79b4b25b 100644
--- a/crates/paimon/src/spec/predicate.rs
+++ b/crates/paimon/src/spec/predicate.rs
@@ -1206,13 +1206,51 @@ pub enum Transform {
Concat(Vec<TransformInput>),
/// Concatenation where the first input is the separator.
ConcatWs(Vec<TransformInput>),
+ /// SQL substring with one-based, zero or negative positions.
+ Substring(Vec<TransformInput>),
+ /// Remove whole characters at the requested ends.
+ Trim(Vec<TransformInput>, TrimFlag),
+ /// Number of Unicode characters, as INT.
+ Length(TransformInput),
+ /// Number of UTF-8 bits, as INT.
+ BitLength(TransformInput),
+ /// SQL LPAD/RPAD.
+ Pad(Vec<TransformInput>, PadDirection),
+ /// Map or delete Unicode characters.
+ Translate(Vec<TransformInput>),
+ /// SQL overlay with optional replacement length.
+ Overlay(Vec<TransformInput>),
}
-/// One input of a string [`Transform`].
+/// The ends selected by Java's TRIM transform.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TrimFlag {
+ /// Remove leading characters.
+ Leading,
+ /// Remove trailing characters.
+ Trailing,
+ /// Remove characters at both ends.
+ Both,
+}
+
+/// The side selected by Java's PAD transform.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PadDirection {
+ /// Prefix the padding.
+ Left,
+ /// Append the padding.
+ Right,
+}
+
+/// One input of a [`Transform`].
#[derive(Debug, Clone, PartialEq)]
pub enum TransformInput {
/// A string literal (`None` = a JSON `null` input, i.e. SQL NULL).
Literal(Option<String>),
+ /// An integer position or length, checked as a Java int when evaluated.
+ Integer(Option<i64>),
+ /// SUBSTRING parses string and large integral positions only after NULL
checks.
+ IntegerText(String),
/// A field reference.
Field(usize),
}
@@ -1233,11 +1271,8 @@ impl Transform {
Transform::FieldRef(index) | Transform::Cast(index, _) => {
out.insert(*index);
}
- Transform::Upper(inputs)
- | Transform::Lower(inputs)
- | Transform::Concat(inputs)
- | Transform::ConcatWs(inputs) => {
- for input in inputs {
+ _ => {
+ for input in self.inputs() {
if let TransformInput::Field(index) = input {
out.insert(*index);
}
@@ -1245,6 +1280,22 @@ impl Transform {
}
}
}
+
+ pub(crate) fn inputs(&self) -> &[TransformInput] {
+ match self {
+ Self::Upper(inputs)
+ | Self::Lower(inputs)
+ | Self::Concat(inputs)
+ | Self::ConcatWs(inputs)
+ | Self::Substring(inputs)
+ | Self::Trim(inputs, _)
+ | Self::Pad(inputs, _)
+ | Self::Translate(inputs)
+ | Self::Overlay(inputs) => inputs,
+ Self::Length(input) | Self::BitLength(input) =>
std::slice::from_ref(input),
+ _ => &[],
+ }
+ }
}
fn parse_rest_transform(value: &serde_json::Value, fields: &[DataField]) ->
Result<Transform> {
@@ -1273,6 +1324,7 @@ fn parse_rest_transform(value: &serde_json::Value,
fields: &[DataField]) -> Resu
serde_json::Value::Null => Ok(TransformInput::Literal(None)),
serde_json::Value::Object(_) => {
let index = field_ref_index(input)?;
+ check_input_type(input, false)?;
match fields[index].data_type() {
DataType::Char(_) | DataType::VarChar(_) => {
Ok(TransformInput::Field(index))
@@ -1319,10 +1371,171 @@ fn parse_rest_transform(value: &serde_json::Value,
fields: &[DataField]) -> Resu
"LOWER" => Ok(Transform::Lower(string_inputs(1, 1)?)),
"CONCAT" => Ok(Transform::Concat(string_inputs(0, usize::MAX)?)),
"CONCAT_WS" => Ok(Transform::ConcatWs(string_inputs(2, usize::MAX)?)),
+ "SUBSTRING" => Ok(Transform::Substring(transform_inputs(
+ value,
+ fields,
+ 2,
+ 3,
+ &[1, 2],
+ true,
+ )?)),
+ "TRIM" => {
+ let flag = match str_prop(value, "trimFlag")? {
+ "LEADING" => TrimFlag::Leading,
+ "TRAILING" => TrimFlag::Trailing,
+ "BOTH" => TrimFlag::Both,
+ flag => return Err(rest_json_err(format!("invalid trimFlag
`{flag}`"))),
+ };
+ Ok(Transform::Trim(string_inputs(1, 2)?, flag))
+ }
+ "LENGTH" => Ok(Transform::Length(string_inputs(1, 1)?.remove(0))),
+ "BIT_LENGTH" => Ok(Transform::BitLength(string_inputs(1,
1)?.remove(0))),
+ "PAD" => {
+ let direction = match str_prop(value, "direction")? {
+ "LEFT" => PadDirection::Left,
+ "RIGHT" => PadDirection::Right,
+ direction => {
+ return Err(rest_json_err(format!(
+ "invalid PAD direction `{direction}`"
+ )))
+ }
+ };
+ Ok(Transform::Pad(
+ transform_inputs(value, fields, 3, 3, &[1], false)?,
+ direction,
+ ))
+ }
+ "TRANSLATE" => Ok(Transform::Translate(string_inputs(3, 3)?)),
+ "OVERLAY" => Ok(Transform::Overlay(transform_inputs(
+ value,
+ fields,
+ 3,
+ 4,
+ &[2, 3],
+ false,
+ )?)),
other => Err(rest_json_err(format!("transform `{other}`"))),
}
}
+/// Bind mixed string/integer inputs by name to the actual table schema.
+fn transform_inputs(
+ value: &serde_json::Value,
+ fields: &[DataField],
+ min: usize,
+ max: usize,
+ integers: &[usize],
+ string_positions: bool,
+) -> Result<Vec<TransformInput>> {
+ let inputs = value
+ .get("inputs")
+ .and_then(serde_json::Value::as_array)
+ .ok_or_else(|| rest_json_err("transform without inputs"))?;
+ if !(min..=max).contains(&inputs.len()) {
+ return Err(rest_json_err(format!(
+ "transform expects {min}..={max} inputs, got {}",
+ inputs.len()
+ )));
+ }
+ inputs
+ .iter()
+ .enumerate()
+ .map(|(slot, input)| {
+ let integer = integers.contains(&slot);
+ match input {
+ serde_json::Value::Null => Ok(if integer {
+ TransformInput::Integer(None)
+ } else {
+ TransformInput::Literal(None)
+ }),
+ serde_json::Value::String(text) if !integer => {
+ Ok(TransformInput::Literal(Some(text.clone())))
+ }
+ serde_json::Value::String(text) if string_positions => {
+ Ok(TransformInput::IntegerText(text.clone()))
+ }
+ serde_json::Value::Number(number) if integer => {
+ // arbitrary_precision retains the lexical integer even
beyond
+ // i64. SUBSTRING must not convert it before row NULL
checks.
+ let text = number.to_string();
+ if text
+ .strip_prefix('-')
+ .unwrap_or(&text)
+ .bytes()
+ .all(|c| c.is_ascii_digit())
+ {
+ match number.as_i64() {
+ Some(number) =>
Ok(TransformInput::Integer(Some(number))),
+ None if string_positions =>
Ok(TransformInput::IntegerText(text)),
+ None => Err(rest_json_err("position exceeds BIGINT
range")),
+ }
+ } else {
+ Err(rest_json_err("position must be an integral
number"))
+ }
+ }
+ serde_json::Value::Object(_) => {
+ let name = str_prop(input, "name")?;
+ if !string_positions {
+ check_input_type(input, integer)?;
+ } else if let Some(stored) =
input.get("type").filter(|value| !value.is_null())
+ {
+ serde_json::from_value::<DataType>(stored.clone())
+ .map_err(rest_json_err)?;
+ }
+ let index = fields
+ .iter()
+ .position(|field| field.name() == name)
+ .ok_or_else(|| rest_json_err(format!("unknown field
`{name}`")))?;
+ let valid = if integer {
+ matches!(
+ fields[index].data_type(),
+ DataType::TinyInt(_)
+ | DataType::SmallInt(_)
+ | DataType::Int(_)
+ | DataType::BigInt(_)
+ )
+ } else {
+ matches!(
+ fields[index].data_type(),
+ DataType::Char(_) | DataType::VarChar(_)
+ )
+ };
+ if !valid {
+ return Err(rest_json_err(format!(
+ "invalid type for transform input `{name}`"
+ )));
+ }
+ Ok(TransformInput::Field(index))
+ }
+ _ => Err(rest_json_err(format!("invalid transform input
{input}"))),
+ }
+ })
+ .collect()
+}
+
+/// Java checks the wire FieldRef family before replacing it from the schema.
+/// SUBSTRING deliberately allows arbitrary stored families and validates the
+/// rebound field at execution instead.
+fn check_input_type(input: &serde_json::Value, integer: bool) -> Result<()> {
+ let stored: DataType = input
+ .get("type")
+ .and_then(|value| serde_json::from_value(value.clone()).ok())
+ .ok_or_else(|| rest_json_err("transform field without a parseable
type"))?;
+ let valid = if integer {
+ matches!(
+ stored,
+ DataType::TinyInt(_) | DataType::SmallInt(_) | DataType::Int(_) |
DataType::BigInt(_)
+ )
+ } else {
+ matches!(stored, DataType::Char(_) | DataType::VarChar(_))
+ };
+ if valid {
+ Ok(())
+ } else {
+ Err(rest_json_err("invalid stored type for transform field"))
+ }
+}
+
// ---------------------------------------------------------------------------
// PredicateBuilder
// ---------------------------------------------------------------------------
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 2377b4be..69b13bbe 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -178,6 +178,36 @@ impl TableSchema {
/// fail, so the stored value wins instead, and an absent one leaves the
/// default standing.
pub fn copy_with_options(&self, mut extra: HashMap<String, String>) ->
Self {
+ self.retain_immutable_options(&mut extra);
+ let mut new_schema = self.clone();
+ new_schema.options.extend(extra);
+ new_schema
+ }
+
+ /// REST returns a Schema without its creation timestamp. Independently
+ /// loaded copies can have different local timestamps while describing the
+ /// same schema. Options are compared separately when copying a handle.
+ pub(crate) fn has_same_definition(&self, other: &Self) -> bool {
+ self.version == other.version
+ && self.id == other.id
+ && self.fields == other.fields
+ && self.highest_field_id == other.highest_field_id
+ && self.partition_keys == other.partition_keys
+ && self.primary_keys == other.primary_keys
+ && self.comment == other.comment
+ }
+
+ /// Replace dynamic options without allowing an override of stored
immutable
+ /// options. Unlike merging, absence in this map removes a mutable option.
+ pub(crate) fn copy_with_replaced_dynamic_options(
+ &self,
+ mut options: HashMap<String, String>,
+ ) -> Self {
+ self.retain_immutable_options(&mut options);
+ self.copy_with_replaced_options(options)
+ }
+
+ fn retain_immutable_options(&self, extra: &mut HashMap<String, String>) {
if self.core_options().query_auth_enabled() {
extra.insert(QUERY_AUTH_ENABLED_OPTION.to_string(),
"true".to_string());
}
@@ -194,9 +224,6 @@ impl TableSchema {
None => extra.remove(key),
};
}
- let mut new_schema = self.clone();
- new_schema.options.extend(extra);
- new_schema
}
/// Create a copy of this schema with the options replaced entirely,
diff --git a/crates/paimon/src/table/goldens/split_v1_query_auth.bin
b/crates/paimon/src/table/goldens/split_v1_query_auth.bin
new file mode 100644
index 00000000..3978a439
Binary files /dev/null and
b/crates/paimon/src/table/goldens/split_v1_query_auth.bin differ
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index ff1db75b..bd8f2cd5 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -645,11 +645,33 @@ impl Table {
/// and does not perform time travel. Unlike that constructor, it retains
the
/// FileIO provider, REST environment and identity of this table. The
branch
/// selects its metadata namespace without loading a different schema.
+ /// An options-only replacement of the catalog-loaded schema keeps its REST
+ /// authorization session and the stored immutable options.
/// Any cached time-travel resolution is discarded: subsequent scans
resolve
/// the supplied options without replacing the supplied fields.
pub fn copy_with_resolved_schema(&self, schema: TableSchema, branch: &str)
-> Result<Self> {
schema.validate_resolved_structure()?;
validate_branch_name(branch)?;
+ // An options-only copy still reads the exact schema loaded by the
+ // catalog. Preserve its session through the existing dynamic-option
+ // path; supplied fields, IDs, keys, comments and branch must all
match.
+ // REST synthesizes creation times independently in different clients.
+ if self.query_auth_session.is_some()
+ && !self.time_traveled
+ && self.travel_snapshot.is_none()
+ && branch == self.branch
+ && schema.has_same_definition(&self.schema)
+ {
+ return Ok(Self {
+ schema: self
+ .schema
+
.copy_with_replaced_dynamic_options(schema.options().clone()),
+ time_traveled: false,
+ travel_snapshot: None,
+ applied_dynamic_option_keys: HashSet::new(),
+ ..self.clone()
+ });
+ }
let schema_manager = SchemaManager::new(self.file_io.clone(),
self.location.clone());
let schema_manager = if branch == DEFAULT_MAIN_BRANCH {
schema_manager
diff --git a/crates/paimon/src/table/query_auth.rs
b/crates/paimon/src/table/query_auth.rs
index 3ae7d349..3de5625a 100644
--- a/crates/paimon/src/table/query_auth.rs
+++ b/crates/paimon/src/table/query_auth.rs
@@ -24,12 +24,20 @@ pub(crate) use rules::{filter_batch, mask_batch,
mask_inputs, ColumnMask, Rules}
/// The server's answer for one user on one table; `session` ties it to the
/// handle that asked, as the response names no table or user.
-#[derive(Debug, PartialEq)]
+#[derive(Debug)]
pub(crate) struct QueryAuthGrant {
session: u64,
/// The columns the request asked about; `None` asked about the whole
table.
select: Option<Vec<String>>,
rules: Rules,
+ /// Original wire rules for Java QueryAuthSplit metadata transport.
+ response: Option<AuthTableQueryResponse>,
+}
+
+impl PartialEq for QueryAuthGrant {
+ fn eq(&self, other: &Self) -> bool {
+ self.session == other.session && self.select == other.select &&
self.rules == other.rules
+ }
}
impl QueryAuthGrant {
@@ -40,7 +48,9 @@ impl QueryAuthGrant {
select: Option<Vec<String>>,
fields: &[crate::spec::DataField],
) -> crate::Result<Self> {
- Ok(Self::new(session, select, Rules::parse(response, fields)?))
+ let mut grant = Self::new(session, select, Rules::parse(response,
fields)?);
+ grant.response = Some(response.clone());
+ Ok(grant)
}
pub(crate) fn new(session: u64, select: Option<Vec<String>>, rules: Rules)
-> Self {
@@ -48,9 +58,14 @@ impl QueryAuthGrant {
session,
select,
rules,
+ response: None,
}
}
+ pub(crate) fn response(&self) -> Option<&AuthTableQueryResponse> {
+ self.response.as_ref()
+ }
+
/// No row filter and no masking.
pub(crate) fn is_unrestricted(&self) -> bool {
self.rules.is_empty()
@@ -405,13 +420,83 @@ mod tests {
}
#[tokio::test]
- async fn test_a_schema_replaced_copy_loses_its_session() {
+ async fn test_options_only_schema_replacement_keeps_its_session() {
let table = crate::table::rest_query_auth_table().await;
assert!(table.query_auth_session().is_some());
+ let mut json = serde_json::to_value(table.schema()).unwrap();
+ json["timeMillis"] = serde_json::json!(1);
let copy = table
- .copy_with_resolved_schema(table.schema().clone(), "main")
+ .copy_with_resolved_schema(serde_json::from_value(json).unwrap(),
"main")
.unwrap();
- assert!(copy.query_auth_session().is_none());
+ assert_eq!(copy.query_auth_session(), table.query_auth_session());
+ let table = table.copy_with_options(std::collections::HashMap::from([
+ ("scan.snapshot-id".into(), "7".into()),
+ ("read.batch-size".into(), "99".into()),
+ ]));
+ let schema = table
+ .schema()
+ .copy_with_replaced_options(std::collections::HashMap::from([
+ ("query-auth.enabled".into(), "false".into()),
+ ("type".into(), "format".into()),
+ ("data-file.path-directory".into(), "/tmp/other".into()),
+ ]));
+ let copy = table.copy_with_resolved_schema(schema, "main").unwrap();
+ assert_eq!(copy.query_auth_session(), table.query_auth_session());
+ assert!(copy.schema().core_options().query_auth_enabled());
+ for key in [
+ "scan.snapshot-id",
+ "read.batch-size",
+ "type",
+ "data-file.path-directory",
+ ] {
+ assert!(!copy.schema().options().contains_key(key), "{key}");
+ }
+ assert!(!copy.reads_another_schema().unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_changed_schema_or_branch_loses_its_session() {
+ let table = crate::table::rest_query_auth_table().await;
+ let mut json = serde_json::to_value(table.schema()).unwrap();
+ json["id"] = serde_json::json!(table.schema().id() + 1);
+ let changed = serde_json::from_value(json).unwrap();
+ assert!(table
+ .copy_with_resolved_schema(changed, "main")
+ .unwrap()
+ .query_auth_session()
+ .is_none());
+ assert!(table
+ .copy_with_resolved_schema(table.schema().clone(), "test")
+ .unwrap()
+ .query_auth_session()
+ .is_none());
+ let assembled = crate::table::query_auth_table();
+ assert!(assembled
+ .copy_with_resolved_schema(assembled.schema().clone(), "main")
+ .unwrap()
+ .query_auth_session()
+ .is_none());
+ }
+
+ #[tokio::test]
+ async fn
test_resolved_schema_copy_cannot_reauthorize_a_historical_handle() {
+ let table = crate::table::rest_query_auth_table().await;
+ let grant = super::QueryAuthGrant::new(
+ table.query_auth_session().unwrap(),
+ None,
+ Default::default(),
+ );
+ let mut travelled = table.clone();
+ travelled.time_traveled = true;
+ let pinned = table.copy_with_pinned_snapshot(None);
+ for view in [travelled, pinned] {
+ let schema = view
+ .schema()
+ .copy_with_replaced_options(table.schema().options().clone());
+ let copy = view.copy_with_resolved_schema(schema, "main").unwrap();
+ assert!(copy.query_auth_session().is_none());
+ assert!(!grant.matches_table(©));
+ }
}
#[tokio::test]
diff --git a/crates/paimon/src/table/query_auth/rules.rs
b/crates/paimon/src/table/query_auth/rules.rs
index 08650011..d1e71ac0 100644
--- a/crates/paimon/src/table/query_auth/rules.rs
+++ b/crates/paimon/src/table/query_auth/rules.rs
@@ -391,6 +391,9 @@ pub(crate) fn mask_batch(
| 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 = cast_to(&masked, &target_type)?;
for position in targets {
@@ -408,6 +411,8 @@ pub(crate) fn mask_batch(
fn mask_can_be_null(transform: &Transform, fields: &[DataField]) -> bool {
let nullable = |input: &TransformInput| match input {
TransformInput::Literal(literal) => literal.is_none(),
+ TransformInput::Integer(value) => value.is_none(),
+ TransformInput::IntegerText(_) => false,
TransformInput::Field(index) =>
fields[*index].data_type().is_nullable(),
};
match transform {
@@ -415,11 +420,9 @@ fn mask_can_be_null(transform: &Transform, fields:
&[DataField]) -> bool {
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),
+ _ => transform.inputs().iter().any(nullable),
}
}
@@ -431,10 +434,8 @@ fn mask_output_type(transform: &Transform, fields:
&[DataField]) -> Option<arrow
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),
+ Transform::Length(_) | Transform::BitLength(_) =>
Some(arrow_schema::DataType::Int32),
+ _ => Some(arrow_schema::DataType::Utf8),
}
}
@@ -536,6 +537,9 @@ fn string_mask(
.iter()
.map(|i| match i {
TransformInput::Literal(literal) =>
Ok(StringInput::Literal(literal.as_deref())),
+ TransformInput::Integer(_) | TransformInput::IntegerText(_) => {
+ Err(unsupported("string mask has an integer input"))
+ }
TransformInput::Field(index) => Ok(StringInput::Column(
cast_to(&input(*index)?, &arrow_schema::DataType::Utf8)?
.as_string::<i32>()
@@ -651,6 +655,50 @@ mod tests {
DataType::VarChar(VarCharType::string_type())
}
+ #[test]
+ fn extended_string_masks_match_java_character_semantics() {
+ let fields = vec![DataField::new(0, "value".into(), string())];
+ let source = batch(
+ &fields,
+ vec![Arc::new(StringArray::from(vec![
+ Some("甲🙂乙"),
+ Some("abcabc"),
+ Some(""),
+ None,
+ ]))],
+ );
+ for (rule, expected) in [
+ (
+
r#"{"name":"SUBSTRING","inputs":[{"name":"value","type":"STRING"},2,2]}"#,
+ vec![Some("🙂乙"), Some("bc"), Some(""), None],
+ ),
+ (
+
r#"{"name":"TRIM","trimFlag":"BOTH","inputs":[{"name":"value","type":"STRING"},"甲乙ac"]}"#,
+ vec![Some("🙂"), Some("bcab"), Some(""), None],
+ ),
+ (
+
r#"{"name":"PAD","direction":"LEFT","inputs":[{"name":"value","type":"STRING"},5,"🙂x"]}"#,
+ vec![Some("🙂x甲🙂乙"), Some("abcab"), Some("🙂x🙂x🙂"), None],
+ ),
+ (
+
r#"{"name":"TRANSLATE","inputs":[{"name":"value","type":"STRING"},"甲🙂乙aba","XY"]}"#,
+ vec![Some("XY"), Some("cc"), Some(""), None],
+ ),
+ (
+
r#"{"name":"OVERLAY","inputs":[{"name":"value","type":"STRING"},"x",2,1]}"#,
+ vec![Some("甲x乙"), Some("axcabc"), Some("x"), None],
+ ),
+ ] {
+ let rules = parse(&[], &[("value", rule)], &fields).unwrap();
+ let result = mask_batch(&source, &rules.masks, &fields,
&fields).unwrap();
+ assert_eq!(
+ result.column(0).as_string::<i32>(),
+ &StringArray::from(expected),
+ "{rule}"
+ );
+ }
+ }
+
fn batch(fields: &[DataField], columns: Vec<ArrayRef>) -> RecordBatch {
let schema: Vec<Field> = fields
.iter()
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index 11d6bc3f..593f432c 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -771,8 +771,8 @@ impl DataSplit {
DataSplitBuilder::new()
}
- /// The Java-compatible frames have no field for the marker, so a reader
would
- /// rebuild the split without it. Serde keeps it; these two must refuse.
+ /// A bare DataSplit frame has no authorization marker and must refuse.
+ /// SplitSerializer can instead transport the Java QueryAuthSplit wrapper.
fn ensure_serializable_without_grant(&self) -> crate::Result<()> {
if self.query_auth_required {
return Err(crate::table::query_auth::unsupported(
@@ -940,7 +940,8 @@ impl DataSplit {
/// Java `SplitSerializer#serialize` frame: magic + version + type id +
body. The cross-language
/// entry point. A plain split serializes as `DataSplit` (type 1); a split
carrying row ranges as
- /// `IndexedSplit` (type 3) wrapping the DataSplit body plus the ranges.
Byte-compatible with
+ /// `IndexedSplit` (type 3) wrapping the DataSplit body plus the ranges.
Authorized metadata
+ /// uses `QueryAuthSplit` (type 5), retaining the complete captured REST
response. Byte-compatible with
/// `compatibility/split-v1-data` / `split-v1-indexed`.
pub fn serialize_split_v1(&self) -> crate::Result<Vec<u8>> {
self.serialize_split_v1_with_scores(None)
@@ -949,6 +950,29 @@ impl DataSplit {
/// Serialize aligned vector scores using Java's IndexedSplit wire format.
/// Scores belong to the selection, in ascending expanded-range order.
pub fn serialize_split_v1_with_scores(&self, scores: Option<&[f32]>) ->
crate::Result<Vec<u8>> {
+ if self.query_auth_required {
+ let response = self
+ .query_auth_grant
+ .as_ref()
+ .and_then(|grant| grant.response())
+ .ok_or_else(|| {
+ crate::table::query_auth::unsupported(
+ "the split has no live wire authorization",
+ )
+ })?;
+ // The nested DataSplit contains metadata only. The outer frame
+ // carries the rules and cannot decode as an ordinary DataSplit.
+ let mut metadata = self.clone();
+ metadata.query_auth_required = false;
+ metadata.query_auth_grant = None;
+ let mut out = Vec::new();
+ out.extend_from_slice(&SPLIT_SER_MAGIC.to_be_bytes());
+ out.extend_from_slice(&SPLIT_SER_VERSION.to_be_bytes());
+
out.extend_from_slice(&SPLIT_SER_TYPE_QUERY_AUTH_SPLIT.to_be_bytes());
+
out.extend_from_slice(&metadata.serialize_split_v1_with_scores(scores)?);
+ write_auth_response(&mut out, response)?;
+ return Ok(out);
+ }
self.ensure_serializable_without_grant()?;
if let Some(scores) = scores {
let count = self.row_ranges.as_ref().and_then(|ranges| {
@@ -999,42 +1023,68 @@ impl DataSplit {
}
/// Reverse of [`DataSplit::serialize_split_v1`]: parse a Java
`SplitSerializer` frame into a
- /// `DataSplit` (type 1) or an `IndexedSplit` (type 3, which carries
`row_ranges`). Other split
- /// type ids, and IndexedSplit vector scores, are unsupported. Consumes
the entire buffer.
+ /// `DataSplit` (type 1), an `IndexedSplit` (type 3, which carries
`row_ranges`), or a
+ /// `QueryAuthSplit` (type 5, retaining the authorization-required marker
without minting
+ /// a grant). Other type ids, and IndexedSplit vector scores, are
unsupported.
+ /// Consumes the entire buffer.
pub fn deserialize_split_v1(data: &[u8]) -> crate::Result<DataSplit> {
let mut cur = data;
- let magic = read_i64(&mut cur)?;
+ let split = Self::read_split_v1(&mut cur, 0)?;
+ if !cur.is_empty() {
+ return Err(crate::Error::DataInvalid {
+ message: format!("{} trailing bytes after SPLIT_V1 frame",
cur.len()),
+ source: None,
+ });
+ }
+ Ok(split)
+ }
+
+ fn read_split_v1(cur: &mut &[u8], depth: usize) ->
crate::Result<DataSplit> {
+ if depth >= 32 {
+ return Err(crate::Error::DataInvalid {
+ message: "QueryAuthSplit nesting exceeds 32".into(),
+ source: None,
+ });
+ }
+ let magic = read_i64(cur)?;
if magic != SPLIT_SER_MAGIC {
return Err(crate::Error::DataInvalid {
message: format!("invalid SPLIT_V1 magic: {magic:#018x}"),
source: None,
});
}
- let version = read_i32(&mut cur)?;
+ let version = read_i32(cur)?;
if version != SPLIT_SER_VERSION {
return Err(crate::Error::Unsupported {
message: format!("SplitSerializer version {version} not
supported"),
});
}
- let type_id = read_i32(&mut cur)?;
+ let type_id = read_i32(cur)?;
let split = match type_id {
- SPLIT_SER_TYPE_DATA_SPLIT => Self::read_body(&mut cur)?,
+ SPLIT_SER_TYPE_QUERY_AUTH_SPLIT => {
+ let split = Self::read_split_v1(cur, depth + 1)?;
+ read_auth_response(cur)?;
+ // Rules transported in a frame are not a grant minted by this
+ // catalog session. Reading it still requires the original
split.
+ split.planned(None)
+ }
+ SPLIT_SER_TYPE_DATA_SPLIT => Self::read_body(cur)?,
SPLIT_SER_TYPE_INDEXED_SPLIT => {
- let im = read_i64(&mut cur)?;
+ let im = read_i64(cur)?;
if im != INDEXED_SPLIT_MAGIC {
return Err(crate::Error::DataInvalid {
message: format!("invalid IndexedSplit magic:
{im:#018x}"),
source: None,
});
}
- let iv = read_i32(&mut cur)?;
+ let iv = read_i32(cur)?;
if iv != INDEXED_SPLIT_VERSION {
return Err(crate::Error::Unsupported {
message: format!("IndexedSplit version {iv} not
supported"),
});
}
- let body = Self::read_body(&mut cur)?;
- let ranges_n = read_i32(&mut cur)?;
+ let body = Self::read_body(cur)?;
+ let ranges_n = read_i32(cur)?;
if ranges_n < 0 {
return Err(crate::Error::DataInvalid {
message: "negative row_ranges count".into(),
@@ -1045,8 +1095,8 @@ impl DataSplit {
// 16 bytes (two i64), so a crafted huge count cannot
over-reserve.
let mut ranges = Vec::with_capacity((ranges_n as
usize).min(cur.len() / 16));
for _ in 0..ranges_n {
- let from = read_i64(&mut cur)?;
- let to = read_i64(&mut cur)?;
+ let from = read_i64(cur)?;
+ let to = read_i64(cur)?;
if from > to {
return Err(crate::Error::DataInvalid {
message: format!("row range from {from} > to
{to}"),
@@ -1055,7 +1105,7 @@ impl DataSplit {
}
ranges.push(RowRange::new(from, to));
}
- let scores_flag = read_u8(&mut cur)?;
+ let scores_flag = read_u8(cur)?;
if scores_flag != 0 {
return Err(crate::Error::Unsupported {
message: "IndexedSplit scores are not
supported".into(),
@@ -1071,12 +1121,6 @@ impl DataSplit {
});
}
};
- if !cur.is_empty() {
- return Err(crate::Error::DataInvalid {
- message: format!("{} trailing bytes after SPLIT_V1 frame",
cur.len()),
- source: None,
- });
- }
Ok(split)
}
}
@@ -1094,6 +1138,87 @@ const SPLIT_SER_MAGIC: i64 = 0x53504C49545F5631; //
"SPLIT_V1"
const SPLIT_SER_VERSION: i32 = 1;
const SPLIT_SER_TYPE_DATA_SPLIT: i32 = 1;
const SPLIT_SER_TYPE_INDEXED_SPLIT: i32 = 3;
+const SPLIT_SER_TYPE_QUERY_AUTH_SPLIT: i32 = 5;
+
+fn write_auth_response(
+ out: &mut Vec<u8>,
+ response: &crate::api::AuthTableQueryResponse,
+) -> crate::Result<()> {
+ fn string(out: &mut Vec<u8>, value: &str) -> crate::Result<()> {
+ let length = i32::try_from(value.len()).map_err(|e|
crate::Error::DataInvalid {
+ message: e.to_string(),
+ source: None,
+ })?;
+ out.extend_from_slice(&length.to_be_bytes());
+ out.extend_from_slice(value.as_bytes());
+ Ok(())
+ }
+ out.push(1);
+ match &response.filter {
+ None => out.push(0),
+ Some(filters) => {
+ out.push(1);
+ out.extend_from_slice(&(filters.len() as i32).to_be_bytes());
+ for filter in filters {
+ string(out, filter)?;
+ }
+ }
+ }
+ match &response.column_masking {
+ None => out.push(0),
+ Some(masks) => {
+ out.push(1);
+ out.extend_from_slice(&(masks.len() as i32).to_be_bytes());
+ let mut masks: Vec<_> = masks.iter().collect();
+ // Java's TreeMap compares UTF-16 units, including supplementary
keys.
+ masks.sort_by(|(a, _), (b, _)|
a.encode_utf16().cmp(b.encode_utf16()));
+ for (name, mask) in masks {
+ string(out, name)?;
+ string(out, mask)?;
+ }
+ }
+ }
+ Ok(())
+}
+
+/// Consume wire rules without creating a trusted grant from untrusted bytes.
+fn read_auth_response(cur: &mut &[u8]) -> crate::Result<()> {
+ fn string(cur: &mut &[u8]) -> crate::Result<()> {
+ let length = read_i32(cur)?;
+ if length < 0 {
+ return Err(crate::Error::DataInvalid {
+ message: "NULL authorization rule in QueryAuthSplit".into(),
+ source: None,
+ });
+ }
+ let bytes = take(cur, length as usize)?;
+ std::str::from_utf8(bytes).map_err(|e| crate::Error::DataInvalid {
+ message: e.to_string(),
+ source: None,
+ })?;
+ Ok(())
+ }
+ if read_u8(cur)? == 0 {
+ return Ok(());
+ }
+ for multiplier in [1, 2] {
+ if read_u8(cur)? != 0 {
+ let count = read_i32(cur)?;
+ if count < 0 || count as usize > cur.len() / (4 * multiplier) {
+ return Err(crate::Error::DataInvalid {
+ message: "invalid QueryAuthSplit rule count".into(),
+ source: None,
+ });
+ }
+ for _ in 0..count {
+ for _ in 0..multiplier {
+ string(cur)?;
+ }
+ }
+ }
+ }
+ Ok(())
+}
/// Java `IndexedSplit#MAGIC` / `VERSION`.
const INDEXED_SPLIT_MAGIC: i64 = -938472394838495695;
@@ -2194,6 +2319,105 @@ mod tests {
}
}
+ #[test]
+ fn query_auth_frame_matches_java_and_does_not_restore_a_live_grant() {
+ let response = crate::api::AuthTableQueryResponse {
+ filter: Some(vec!["row甲🙂\0".into()]),
+ column_masking: Some(std::collections::HashMap::from([
+ ("\u{e000}".into(), "bmp".into()),
+ ("𐀀".into(), "supplementary".into()),
+ ])),
+ };
+ // Captured from Java QueryAuthSplit.serialize: UTF-8 with int length,
+ // NUL as one byte, and keys sorted by UTF-16 code units.
+ let expected =
hex::decode("0101000000010000000b726f77e794b2f09f998200010000000200000004f09080800000000d737570706c656d656e7461727900000003ee808000000003626d70").unwrap();
+ let mut tail = Vec::new();
+ write_auth_response(&mut tail, &response).unwrap();
+ assert_eq!(tail, expected);
+ let grant =
+
QueryAuthGrant::parse(&crate::api::AuthTableQueryResponse::default(), 1, None,
&[])
+ .unwrap();
+ // Use a valid parsed policy for the live frame. The arbitrary strings
+ // above test transport independently of grant validation.
+ let mut frame_tail = Vec::new();
+ write_auth_response(&mut frame_tail,
grant.response().unwrap()).unwrap();
+ let ordinary = v1_data_split_builder().build().unwrap();
+ let marked = ordinary.clone().planned(Some(Arc::new(grant)));
+ assert!(marked.serialize().is_err());
+ let frame = marked.serialize_split_v1().unwrap();
+ let mut expected_frame = Vec::new();
+ expected_frame.extend_from_slice(&SPLIT_SER_MAGIC.to_be_bytes());
+ expected_frame.extend_from_slice(&SPLIT_SER_VERSION.to_be_bytes());
+
expected_frame.extend_from_slice(&SPLIT_SER_TYPE_QUERY_AUTH_SPLIT.to_be_bytes());
+
expected_frame.extend_from_slice(&ordinary.serialize_split_v1().unwrap());
+ expected_frame.extend_from_slice(&frame_tail);
+ assert_eq!(frame, expected_frame);
+ let copy = DataSplit::deserialize_split_v1(&frame).unwrap();
+ assert!(copy.query_auth_required());
+ assert!(copy.query_auth_grant().is_none());
+ assert_eq!(copy.data_files(), ordinary.data_files());
+ assert!(copy.serialize().is_err());
+ assert!(copy.serialize_split_v1().is_err());
+ }
+
+ #[test]
+ fn deserialize_java_query_auth_golden_keeps_the_required_marker() {
+ // Java SplitSerializer(QueryAuthSplit(DataSplit v9, policy)) with BMP,
+ // supplementary and NUL text in its policy strings.
+ let split =
+
DataSplit::deserialize_split_v1(include_bytes!("goldens/split_v1_query_auth.bin"))
+ .unwrap();
+ assert_eq!(split.bucket(), 3);
+ assert_eq!(split.data_files().len(), 2);
+ assert!(split.query_auth_required());
+ assert!(split.query_auth_grant().is_none());
+ assert!(split.serialize_split_v1().is_err());
+ }
+
+ #[test]
+ fn query_auth_frame_bounds_policy_lengths_and_nesting() {
+ let ordinary = v1_data_split_builder()
+ .build()
+ .unwrap()
+ .serialize_split_v1()
+ .unwrap();
+ let mut prefix = Vec::new();
+ prefix.extend_from_slice(&SPLIT_SER_MAGIC.to_be_bytes());
+ prefix.extend_from_slice(&SPLIT_SER_VERSION.to_be_bytes());
+
prefix.extend_from_slice(&SPLIT_SER_TYPE_QUERY_AUTH_SPLIT.to_be_bytes());
+ let mut frame = prefix.clone();
+ frame.extend_from_slice(&ordinary);
+ let policy_start = frame.len();
+ let response = crate::api::AuthTableQueryResponse {
+ filter: Some(vec!["甲".repeat(30000)]),
+ column_masking: Some(Default::default()),
+ };
+ write_auth_response(&mut frame, &response).unwrap();
+ assert!(DataSplit::deserialize_split_v1(&frame)
+ .unwrap()
+ .query_auth_required());
+ for tail in [
+ vec![1, 0],
+ vec![1, 1, 0x7f, 0xff, 0xff, 0xff],
+ vec![1, 1, 0xff, 0xff, 0xff, 0xff],
+ vec![1, 1, 0, 0, 0, 1, 0xff, 0xff, 0xff, 0xff, 0],
+ ] {
+ frame.truncate(policy_start);
+ frame.extend_from_slice(&tail);
+ assert!(DataSplit::deserialize_split_v1(&frame).is_err());
+ }
+ let mut nested = prefix.repeat(33);
+ nested.extend_from_slice(&ordinary);
+ nested.extend_from_slice(&[0; 33]);
+ assert!(DataSplit::deserialize_split_v1(&nested).is_err());
+ frame.truncate(policy_start);
+ frame.extend_from_slice(&[0, 0]);
+ assert!(
+ DataSplit::deserialize_split_v1(&frame).is_err(),
+ "trailing bytes"
+ );
+ }
+
#[test]
fn deserialize_split_v1_rejects_huge_ranges_count_without_aborting() {
let split = DataSplitBuilder::new()
