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 69118de8 feat(auth): authorize query-auth reads and carry the grant on
the split (#758)
69118de8 is described below
commit 69118de8090c9fa9063569f7c640deedc8bd60a1
Author: Jiajia Li <[email protected]>
AuthorDate: Fri Oct 2 10:10:27 2026 +0800
feat(auth): authorize query-auth reads and carry the grant on the split
(#758)
---
.../datafusion/src/partition_count_pushdown.rs | 64 ++-
crates/paimon/src/api/api_response.rs | 1 +
crates/paimon/src/spec/core_options.rs | 2 +-
crates/paimon/src/spec/schema.rs | 29 +-
crates/paimon/src/table/audit_log_table/read.rs | 7 +-
crates/paimon/src/table/format_table_read.rs | 10 +-
crates/paimon/src/table/mod.rs | 108 +++++
crates/paimon/src/table/pk_vector_scan.rs | 29 ++
crates/paimon/src/table/query_auth.rs | 511 +++++++++++++++++++++
crates/paimon/src/table/read_builder.rs | 56 ++-
crates/paimon/src/table/rest_env.rs | 66 ++-
crates/paimon/src/table/source.rs | 77 ++++
crates/paimon/src/table/table_commit.rs | 50 ++
crates/paimon/src/table/table_read.rs | 468 ++++++++++++++++++-
crates/paimon/src/table/table_scan.rs | 147 +++++-
crates/paimon/tests/mock_server.rs | 152 +++++-
crates/paimon/tests/rest_catalog_test.rs | 403 ++++++++++++++++
17 files changed, 2095 insertions(+), 85 deletions(-)
diff --git a/crates/integrations/datafusion/src/partition_count_pushdown.rs
b/crates/integrations/datafusion/src/partition_count_pushdown.rs
index 58aad534..0dc73a53 100644
--- a/crates/integrations/datafusion/src/partition_count_pushdown.rs
+++ b/crates/integrations/datafusion/src/partition_count_pushdown.rs
@@ -296,7 +296,14 @@ impl TableProvider for PartitionRowCountProvider {
.clone();
let table = self.table.clone();
let table = crate::runtime::await_with_runtime(async move {
-
CoreOptions::new(table.schema().options()).ensure_read_authorized()?;
+ // Rules the manifests cannot apply: stay unpinned, so the scan
+ // carries the grant.
+ if CoreOptions::new(table.schema().options())
+ .ensure_read_authorized()
+ .is_err()
+ {
+ return Ok(Some(table));
+ }
if table.travel_snapshot().is_some() {
return Ok(Some(table));
}
@@ -328,6 +335,7 @@ impl TableProvider for PartitionRowCountProvider {
self,
table,
provider_as_source(Arc::new(fallback_provider)),
+ provider_as_source(Arc::new(self.fallback_provider.clone())),
projection.cloned(),
state,
)?);
@@ -352,6 +360,7 @@ struct PartitionRowCountStream {
projection: Option<Vec<usize>>,
output_schema: SchemaRef,
source: Arc<dyn TableSource>,
+ unpinned_source: Arc<dyn TableSource>,
table_name: TableReference,
filters: Vec<Expr>,
state: Arc<SessionState>,
@@ -362,6 +371,7 @@ impl PartitionRowCountStream {
provider: &PartitionRowCountProvider,
table: Option<Table>,
source: Arc<dyn TableSource>,
+ unpinned_source: Arc<dyn TableSource>,
projection: Option<Vec<usize>>,
state: SessionState,
) -> DFResult<Self> {
@@ -374,6 +384,7 @@ impl PartitionRowCountStream {
projection,
output_schema,
source,
+ unpinned_source,
table_name: provider.table_name.clone(),
filters: provider.filters.clone(),
state: Arc::new(state),
@@ -387,13 +398,24 @@ impl PartitionRowCountStream {
let counts = match self.table.clone() {
Some(table) => {
let predicate = self.predicate.clone();
- crate::runtime::await_with_runtime(async move {
+ match crate::runtime::await_with_runtime(async move {
table
.exact_partition_row_counts_with_filter(predicate)
.await
})
.await
- .map_err(to_datafusion_error)?
+ {
+ Ok(counts) => counts,
+ // Refused: only an unpinned scan can authorize.
+ Err(paimon::Error::Unsupported { .. }) => {
+ let plan = crate::runtime::await_with_runtime(
+ self.scan_by_reading(&self.unpinned_source),
+ )
+ .await?;
+ return self.fallback_stream(plan, context);
+ }
+ Err(error) => return Err(to_datafusion_error(error)),
+ }
}
None => Some(Vec::new()),
};
@@ -407,15 +429,9 @@ impl PartitionRowCountStream {
self.table_name,
self.table.as_ref().and_then(Table::travel_snapshot).map(|snapshot|
snapshot.id()),
);
- let plan =
crate::runtime::await_with_runtime(self.scan_by_reading()).await?;
- if plan.schema() != self.output_schema {
- return internal_err!(
- "partition count fallback schema mismatch: expected {:?},
got {:?}",
- self.output_schema,
- plan.schema()
- );
- }
- return datafusion::physical_plan::execute_stream(plan, context);
+ let plan =
+
crate::runtime::await_with_runtime(self.scan_by_reading(&self.source)).await?;
+ return self.fallback_stream(plan, context);
};
// A partition with no surviving rows must not create a GROUP BY key.
@@ -476,8 +492,26 @@ impl PartitionRowCountStream {
}
/// The same rows, computed the ordinary way: count the original scan per
partition.
- async fn scan_by_reading(&self) -> DFResult<Arc<dyn ExecutionPlan>> {
- let source_schema = self.source.schema();
+ fn fallback_stream(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ context: Arc<TaskContext>,
+ ) -> DFResult<SendableRecordBatchStream> {
+ if plan.schema() != self.output_schema {
+ return internal_err!(
+ "partition count fallback schema mismatch: expected {:?}, got
{:?}",
+ self.output_schema,
+ plan.schema()
+ );
+ }
+ datafusion::physical_plan::execute_stream(plan, context)
+ }
+
+ async fn scan_by_reading(
+ &self,
+ source: &Arc<dyn TableSource>,
+ ) -> DFResult<Arc<dyn ExecutionPlan>> {
+ let source_schema = source.schema();
let partition_indices = self
.partition_fields
.iter()
@@ -485,7 +519,7 @@ impl PartitionRowCountStream {
.collect::<Result<Vec<_>, _>>()?;
let scan = LogicalPlan::TableScan(TableScan::try_new(
self.table_name.clone(),
- Arc::clone(&self.source),
+ Arc::clone(source),
Some(partition_indices),
self.filters.clone(),
None,
diff --git a/crates/paimon/src/api/api_response.rs
b/crates/paimon/src/api/api_response.rs
index 943f861c..0a595f02 100644
--- a/crates/paimon/src/api/api_response.rs
+++ b/crates/paimon/src/api/api_response.rs
@@ -568,6 +568,7 @@ impl ListPoliciesResponse {
#[cfg(test)]
mod tests {
+
use super::*;
#[test]
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index d95c9767..8760c454 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -1193,7 +1193,7 @@ impl<'a> CoreOptions<'a> {
.and_then(|v| v.parse().ok())
}
- fn configured_time_travel_selectors(&self) -> Vec<&'static str> {
+ pub(crate) fn configured_time_travel_selectors(&self) -> Vec<&'static str>
{
let mut selectors = Vec::with_capacity(6);
if self.options.contains_key(SCAN_TIMESTAMP_MILLIS_OPTION) {
selectors.push(SCAN_TIMESTAMP_MILLIS_OPTION);
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index c00b3e17..cfe2df5a 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -640,22 +640,27 @@ impl TableSchema {
}
}
-/// Reject column names reserved for system use, mirroring Java
`SpecialFields`:
-/// the five `SYSTEM_FIELD_NAMES` and the `_KEY_` key-field prefix.
+/// Whether `name` is one Paimon reserves for a system column. Java
+/// `SpecialFields.SYSTEM_FIELD_NAMES` plus the `_KEY_` key-field prefix.
+pub(crate) fn is_reserved_system_field_name(name: &str) -> bool {
+ name.starts_with(KEY_FIELD_PREFIX) || SYSTEM_FIELD_NAMES.contains(&name)
+}
+
+// Java SpecialFields.SYSTEM_FIELD_NAMES.
+const SYSTEM_FIELD_NAMES: [&str; 5] = [
+ SEQUENCE_NUMBER_FIELD_NAME,
+ VALUE_KIND_FIELD_NAME,
+ "_LEVEL",
+ ROW_KIND_FIELD_NAME,
+ ROW_ID_FIELD_NAME,
+];
+const KEY_FIELD_PREFIX: &str = "_KEY_";
+
+/// Reject column names reserved for system use, mirroring Java
`SpecialFields`.
///
/// A user column colliding with a system field is otherwise excluded from the
/// physical read and silently filled with the system value.
fn validate_no_reserved_field_names(fields: &[DataField]) -> crate::Result<()>
{
- // Java SpecialFields.SYSTEM_FIELD_NAMES.
- const SYSTEM_FIELD_NAMES: [&str; 5] = [
- SEQUENCE_NUMBER_FIELD_NAME,
- VALUE_KIND_FIELD_NAME,
- "_LEVEL",
- ROW_KIND_FIELD_NAME,
- ROW_ID_FIELD_NAME,
- ];
- const KEY_FIELD_PREFIX: &str = "_KEY_";
-
for field in fields {
let name = field.name();
if name.starts_with(KEY_FIELD_PREFIX) ||
SYSTEM_FIELD_NAMES.contains(&name) {
diff --git a/crates/paimon/src/table/audit_log_table/read.rs
b/crates/paimon/src/table/audit_log_table/read.rs
index d81f1f3d..f63b9c00 100644
--- a/crates/paimon/src/table/audit_log_table/read.rs
+++ b/crates/paimon/src/table/audit_log_table/read.rs
@@ -44,7 +44,9 @@ pub struct AuditLogRead<'a> {
impl<'a> AuditLogRead<'a> {
pub fn new(read: TableRead<'a>) -> crate::Result<Self> {
- read.ensure_query_auth_allowed()?;
+ // Query-auth is decided per split in `to_arrow`; only the type is
known here.
+ crate::spec::CoreOptions::new(read.table().schema().options())
+
.ensure_type_paimon_served(&read.table().identifier().full_name())?;
match read.0 {
TableReadKind::Paimon(read) => Ok(Self { read }),
TableReadKind::Format(_) => Err(crate::Error::Unsupported {
@@ -55,6 +57,9 @@ impl<'a> AuditLogRead<'a> {
/// Reads splits planned by an audit scan, retaining winning retract rows.
pub fn to_arrow(&self, data_splits: &[DataSplit]) ->
crate::Result<ArrowRecordBatchStream> {
+ // The primary-key path below builds its readers directly, so decide
here.
+ self.read
+
.ensure_authorized_by_splits(&self.read.table.schema.core_options(),
data_splits)?;
let output_read_type = self.read.read_type.clone();
if output_read_type
.iter()
diff --git a/crates/paimon/src/table/format_table_read.rs
b/crates/paimon/src/table/format_table_read.rs
index 125a2a07..3eb95274 100644
--- a/crates/paimon/src/table/format_table_read.rs
+++ b/crates/paimon/src/table/format_table_read.rs
@@ -118,7 +118,15 @@ impl<'a> FormatTableRead<'a> {
data_splits: &[DataSplit],
) -> crate::Result<ArrowRecordBatchStream> {
let core_options = self.table.schema().core_options();
- core_options.ensure_read_authorized()?;
+
core_options.ensure_type_paimon_served(&self.table.identifier().full_name())?;
+ // The marker carries the plan's decision.
+ if core_options.query_auth_enabled()
+ || data_splits.iter().any(|split| split.query_auth_required())
+ {
+ return Err(super::query_auth::unsupported(
+ "a format table cannot apply a row filter or column masking",
+ ));
+ }
// Mapping the conjunct onto the data fields drops it, so the read
would
// silently ignore the filter. Guard on the read path, not the builder:
// `TableRead` is public and can be constructed and filtered directly.
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 5bd17835..d9e958b6 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -112,6 +112,7 @@ mod postpone_fixed_bucket_router;
mod postpone_fixed_bucket_write;
mod postpone_fixed_bucket_write_builder;
mod prepared_files;
+mod query_auth;
mod read_builder;
pub mod referenced_files;
pub(crate) mod rest_env;
@@ -235,6 +236,9 @@ pub struct Table {
schema_manager: SchemaManager,
branch: String,
branch_reference: bool,
+ /// Minted only by [`RESTEnv::build_table`]; a [`Table::new`] handle cannot
+ /// replay a grant.
+ query_auth_session: Option<u64>,
rest_env: Option<RESTEnv>,
/// True when this table copy was switched to a historical schema by
/// [`Table::copy_with_time_travel`]. Such a copy is read-only.
@@ -264,6 +268,7 @@ impl Table {
schema_manager,
branch,
branch_reference: false,
+ query_auth_session: None,
rest_env,
time_traveled: false,
travel_snapshot: None,
@@ -304,6 +309,7 @@ impl Table {
schema_manager,
branch,
branch_reference,
+ query_auth_session: None,
rest_env: None,
time_traveled: false,
travel_snapshot: None,
@@ -396,6 +402,75 @@ impl Table {
}
}
+ /// Whether this handle reads a schema other than the one the server rules
+ /// on: a time-travel option, a travelled or branch view, or a `$branch_x`
/
+ /// `$files` name that reads the base table's files.
+ pub(crate) fn reads_another_schema(&self) -> Result<bool> {
+ // Presence only; planning validates the selector after adapting
+ // `scan.version`.
+ let travels = !CoreOptions::new(self.schema.options())
+ .configured_time_travel_selectors()
+ .is_empty();
+ let decorated = self.identifier.branch_name()?.is_some()
+ || self.identifier.system_table_name()?.is_some();
+ Ok(travels || self.time_traveled || self.branch_reference || decorated)
+ }
+
+ /// Whether this user may read this table; `None` when it is not
+ /// `query-auth.enabled`. `query_auth` is the loaded option, as in Java.
+ pub(crate) async fn authorize_read(
+ &self,
+ query_auth: bool,
+ ) -> Result<Option<std::sync::Arc<query_auth::QueryAuthGrant>>> {
+ let local = CoreOptions::new(self.schema.options());
+ let Some(rest_env) = &self.rest_env else {
+ // Only a REST catalog can authorize.
+ return if local.query_auth_enabled() {
+ Err(query_auth::unsupported(
+ "it requires a REST catalog to authorize the query",
+ ))
+ } else {
+ Ok(None)
+ };
+ };
+
+ // No freshness assertion yet — an ordinary table must not inherit one.
+ if !query_auth {
+ return Ok(None);
+ }
+ if self.reads_another_schema()? {
+ return Err(query_auth::unsupported(
+ "a time-travelled or branch read authorizes against the
table's current schema, \
+ which is not the one it reads",
+ ));
+ }
+
+ // Before any RPC: only the catalog mints a session, so an assembled
+ // handle stops here.
+ let session = self.query_auth_session.ok_or_else(|| {
+ query_auth::unsupported("this table handle was assembled rather
than loaded")
+ })?;
+
+ // Naming a system column here would fail the server's column check.
+ let response = rest_env
+ .table_query_auth(self.schema.id(), self.schema.fields(), None)
+ .await?;
+ Ok(Some(std::sync::Arc::new(query_auth::QueryAuthGrant::new(
+ response, session,
+ ))))
+ }
+
+ /// Handed out once per catalog-loaded table; wraps only after 2^64 loads.
+ pub(crate) fn with_query_auth_session(mut self) -> Self {
+ static NEXT: std::sync::atomic::AtomicU64 =
std::sync::atomic::AtomicU64::new(0);
+ self.query_auth_session = Some(NEXT.fetch_add(1,
std::sync::atomic::Ordering::Relaxed));
+ self
+ }
+
+ pub(crate) fn query_auth_session(&self) -> Option<u64> {
+ self.query_auth_session
+ }
+
/// Get the REST environment, if this table was loaded from a REST catalog.
pub fn rest_env(&self) -> Option<&RESTEnv> {
self.rest_env.as_ref()
@@ -505,6 +580,7 @@ impl Table {
schema_manager: self.schema_manager.clone(),
branch: self.branch.clone(),
branch_reference: self.branch_reference,
+ query_auth_session: self.query_auth_session,
rest_env: self.rest_env.clone(),
time_traveled: self.time_traveled,
travel_snapshot: if selector_changed {
@@ -539,6 +615,8 @@ impl Table {
branch_reference: self.branch_reference || branch !=
DEFAULT_MAIN_BRANCH,
time_traveled: false,
travel_snapshot: None,
+ // Not the schema the catalog loaded, so not a handle it
authorizes.
+ query_auth_session: None,
..self.clone()
})
}
@@ -746,6 +824,7 @@ impl Table {
schema_manager,
branch,
branch_reference: true,
+ query_auth_session: self.query_auth_session,
rest_env: self.rest_env.clone(),
time_traveled: false,
travel_snapshot: None,
@@ -780,6 +859,35 @@ pub(crate) fn find_field_id_by_name(fields: &[DataField],
name: &str) -> Option<
fields.iter().find(|f| f.name() == name).map(|f| f.id())
}
+/// A `query-auth.enabled` table wired to its own REST session.
+#[cfg(test)]
+pub(crate) async fn rest_query_auth_table() -> Table {
+ use crate::api::rest_api::RESTApi;
+ use crate::common::{CatalogOptions, Options};
+
+ let mut options = Options::default();
+ options.set(CatalogOptions::URI, "http://127.0.0.1:1";);
+ options.set("token.provider", "bear");
+ options.set("token", "test_token");
+ let api = std::sync::Arc::new(RESTApi::new(options.clone(),
false).await.unwrap());
+ let metadata_cache =
+
crate::io::FileFormatMetadataCacheContext::from_props(options.to_map()).unwrap();
+ let table = query_auth_table();
+ Table {
+ rest_env: Some(RESTEnv::new(
+ table.identifier.clone(),
+ "uuid-1".to_string(),
+ api,
+ options,
+ false,
+ None,
+ metadata_cache,
+ )),
+ ..table
+ }
+ .with_query_auth_session()
+}
+
/// A minimal table with `query-auth.enabled = true`, for the fail-closed read
guard.
#[cfg(test)]
pub(crate) fn query_auth_table() -> Table {
diff --git a/crates/paimon/src/table/pk_vector_scan.rs
b/crates/paimon/src/table/pk_vector_scan.rs
index 8b998bd8..36f460ea 100644
--- a/crates/paimon/src/table/pk_vector_scan.rs
+++ b/crates/paimon/src/table/pk_vector_scan.rs
@@ -439,6 +439,12 @@ fn plan_from_bucket_splits(
"bucket-split planning requires at least one bucket split",
));
}
+ // The split's marker carries the plan's decision, as in `to_arrow`.
+ if splits.iter().any(|s| s.data_split().query_auth_required()) {
+ return Err(crate::table::query_auth::unsupported(
+ "an engine-planned vector split of such a table carries no
authorization",
+ ));
+ }
let mut snapshot_id: Option<i64> = None;
let mut seen_buckets: HashSet<BucketKey> = HashSet::new();
@@ -1314,6 +1320,29 @@ mod tests {
.unwrap_or_default()
}
+ #[test]
+ fn a_marked_engine_split_is_refused() {
+ let data_split = DataSplitBuilder::new()
+ .with_snapshot(11)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path("bucket-0".to_string())
+ .with_total_buckets(1)
+ .with_data_files(vec![dfm("d0", 5, 5, Some(1))])
+ .build()
+ .unwrap()
+ .planned(None);
+ let split = BucketVectorSearchSplit::new_for_test(data_split, vec![],
Default::default());
+ let Err(err) = plan_from_bucket_splits("ivf-pq", 2, None, "/tbl",
false, vec![split])
+ else {
+ panic!("a marked split must not plan")
+ };
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message } if
message.contains("query-auth.enabled")),
+ "{err:?}"
+ );
+ }
+
#[test]
fn plans_the_java_golden_bucket_split() {
let split =
BucketVectorSearchSplit::deserialize(BUCKET_SPLIT_GOLDEN).unwrap();
diff --git a/crates/paimon/src/table/query_auth.rs
b/crates/paimon/src/table/query_auth.rs
new file mode 100644
index 00000000..fb24d3ba
--- /dev/null
+++ b/crates/paimon/src/table/query_auth.rs
@@ -0,0 +1,511 @@
+// 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.
+
+//! What the REST server authorized a user to read from one table.
+
+use crate::api::AuthTableQueryResponse;
+
+/// The server's answer for one user on one table, kept unparsed; `session`
+/// ties it to the handle that asked, as the response names no table or user.
+#[derive(Debug, PartialEq)]
+pub(crate) struct QueryAuthGrant {
+ response: AuthTableQueryResponse,
+ session: u64,
+}
+
+impl QueryAuthGrant {
+ pub(crate) fn new(response: AuthTableQueryResponse, session: u64) -> Self {
+ Self { response, session }
+ }
+
+ /// The only case this client can serve.
+ pub(crate) fn is_unrestricted(&self) -> bool {
+ self.response.is_unrestricted()
+ }
+
+ /// A view of another schema is not the one the server ruled on.
+ pub(crate) fn matches_table(&self, table: &super::Table) -> bool {
+ !table.reads_another_schema().unwrap_or(true)
+ && table.query_auth_session() == Some(self.session)
+ }
+}
+
+/// An older file can name a dropped column; refused rather than scrubbed.
+pub(crate) async fn reject_unauthorized_stats(
+ plan: &super::Plan,
+ current: &crate::spec::TableSchema,
+ schemas: &super::schema_manager::SchemaManager,
+) -> crate::Result<()> {
+ let refuse = |column: &str| {
+ Err(unsupported(&format!(
+ "a data file still carries statistics for '{column}', which the
current schema — the \
+ one the server authorized — does not have"
+ )))
+ };
+ let named = |name: &String| current.fields().iter().any(|f| f.name() ==
name);
+ let mut checked = std::collections::HashSet::new();
+ for split in plan.splits() {
+ for file in split.data_files() {
+ for column in file
+ .value_stats_cols
+ .iter()
+ .chain(file.write_cols.iter())
+ .flatten()
+ {
+ if !named(column) {
+ return refuse(column);
+ }
+ }
+ // The file's schema decides: a name can be re-added under a new
id, and
+ // either list may be absent.
+ if file.schema_id == current.id() ||
!checked.insert(file.schema_id) {
+ continue;
+ }
+ let older = schemas.schema(file.schema_id).await?;
+ if let Some(gone) = older.fields().iter().find(|f| {
+ !current.fields().iter().any(|c| {
+ c.id() == f.id()
+ && c.name() == f.name()
+ && contains(c.data_type(), f.data_type())
+ })
+ }) {
+ return refuse(gone.name());
+ }
+ }
+ }
+ Ok(())
+}
+
+/// Whether `narrow` reads nothing `wide` lacks: nested children match by id
+/// and name, so a `ROW` projection passes and a re-added child does not.
+fn contains(wide: &crate::spec::DataType, narrow: &crate::spec::DataType) ->
bool {
+ use crate::spec::DataType;
+ match (wide, narrow) {
+ (DataType::Row(w), DataType::Row(n)) => n.fields().iter().all(|nf| {
+ w.fields().iter().any(|wf| {
+ wf.id() == nf.id()
+ && wf.name() == nf.name()
+ && contains(wf.data_type(), nf.data_type())
+ })
+ }),
+ (DataType::Array(w), DataType::Array(n)) => contains(w.element_type(),
n.element_type()),
+ (DataType::Multiset(w), DataType::Multiset(n)) => {
+ contains(w.element_type(), n.element_type())
+ }
+ (DataType::Map(w), DataType::Map(n)) => {
+ contains(w.key_type(), n.key_type()) && contains(w.value_type(),
n.value_type())
+ }
+ // A `variant_get` pushdown reads a `VARIANT` column as a `ROW` of
paths.
+ (DataType::Variant(_), DataType::Row(_)) => true,
+ (w, n) => w == n,
+ }
+}
+
+/// A refusal naming the option, so callers never match on prose.
+pub(crate) fn unsupported(reason: &str) -> crate::Error {
+ crate::Error::Unsupported {
+ message: format!(
+ "reading a table with 'query-auth.enabled' = true is not
supported: {reason}"
+ ),
+ }
+}
+
+/// Column permissions cover schema fields only, never `_ROW_ID` and friends.
+pub(crate) fn reject_system_columns<'a>(
+ names: impl IntoIterator<Item = &'a str>,
+) -> crate::Result<()> {
+ for name in names {
+ if crate::spec::is_reserved_system_field_name(name) {
+ return Err(unsupported(&format!(
+ "the system column '{name}' is not one the server can
authorize: column \
+ permissions are granted over table columns"
+ )));
+ }
+ }
+ Ok(())
+}
+
+/// Older files resolve by id, so a non-canonical `(id, name)` pair reads
+/// something no grant covered. System fields have no entry.
+pub(crate) fn reject_noncanonical_fields(
+ read_type: &[crate::spec::DataField],
+ schema_fields: &[crate::spec::DataField],
+) -> crate::Result<()> {
+ for field in read_type {
+ if crate::spec::is_reserved_system_field_name(field.name()) {
+ continue;
+ }
+ // The whole shape: an older field can keep `(id, name)` and carry an
+ // extra nested child.
+ let canonical = schema_fields.iter().any(|f| {
+ f.id() == field.id()
+ && f.name() == field.name()
+ && contains(f.data_type(), field.data_type())
+ });
+ if !canonical {
+ return Err(unsupported(&format!(
+ "'{}' (field id {}) is not a column of the current schema,
which is what the \
+ server authorized",
+ field.name(),
+ field.id()
+ )));
+ }
+ }
+ Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::reject_system_columns;
+ use crate::table::{query_auth_table, rest_query_auth_table};
+
+ #[tokio::test]
+ async fn test_a_grant_is_pinned_to_the_handle_that_obtained_it() {
+ let a = crate::table::rest_query_auth_table().await;
+ let b = crate::table::rest_query_auth_table().await;
+ let grant = super::QueryAuthGrant::new(
+ crate::api::AuthTableQueryResponse::default(),
+ a.query_auth_session().unwrap(),
+ );
+ assert!(grant.matches_table(&a));
+ assert!(
+ !grant.matches_table(&b),
+ "another handle — another principal or another table — must not
reuse it"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_a_time_travel_selector_alone_is_refused() {
+ for selector in [
+ "scan.snapshot-id",
+ "scan.version",
+ "scan.tag-name",
+ "scan.timestamp-millis",
+ "scan.timestamp",
+ "scan.watermark",
+ ] {
+ let table =
+ rest_query_auth_table()
+ .await
+ .copy_with_options(std::collections::HashMap::from([(
+ selector.to_string(),
+ "1".to_string(),
+ )]));
+ assert!(!table.is_time_traveled(), "{selector} sets no flag");
+ let err = table.authorize_read(true).await.unwrap_err();
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("time-travelled or branch read")),
+ "{selector}: {err:?}"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn
test_a_conflicting_selector_pair_is_planning_business_not_authorization() {
+ // `scan.version` is adapted before the one-selector rule, so an
ordinary
+ // table must reach planning rather than fail here.
+ let table = crate::table::Table::new(
+ crate::io::FileIOBuilder::new("file").build().unwrap(),
+ crate::catalog::Identifier::new("default", "plain"),
+ "/tmp/test-plain-selector".to_string(),
+ crate::spec::TableSchema::new(
+ 0,
+ &crate::spec::Schema::builder()
+ .column(
+ "id",
+
crate::spec::DataType::Int(crate::spec::IntType::new()),
+ )
+ .build()
+ .unwrap(),
+ ),
+ None,
+ )
+ .copy_with_options(std::collections::HashMap::from([
+ ("scan.version".to_string(), "1".to_string()),
+ ("scan.snapshot-id".to_string(), "invalid".to_string()),
+ ]));
+ assert!(table.reads_another_schema().unwrap());
+ assert!(table.authorize_read(false).await.unwrap().is_none());
+ }
+
+ #[tokio::test]
+ async fn test_a_schema_replaced_copy_loses_its_session() {
+ let table = crate::table::rest_query_auth_table().await;
+ assert!(table.query_auth_session().is_some());
+ let copy = table
+ .copy_with_resolved_schema(table.schema().clone(), "main")
+ .unwrap();
+ assert!(copy.query_auth_session().is_none());
+ }
+
+ #[tokio::test]
+ async fn test_a_grant_does_not_cross_into_a_travelled_or_branch_view() {
+ let table = crate::table::rest_query_auth_table().await;
+ let grant = super::QueryAuthGrant::new(
+ crate::api::AuthTableQueryResponse::default(),
+ table.query_auth_session().unwrap(),
+ );
+ assert!(grant.matches_table(&table));
+
+ let mut travelled =
table.copy_with_options(std::collections::HashMap::new());
+ travelled.time_traveled = true;
+ assert!(
+ !grant.matches_table(&travelled),
+ "an older schema is not the one the server ruled on"
+ );
+ let selected =
table.copy_with_options(std::collections::HashMap::from([(
+ "scan.snapshot-id".to_string(),
+ "1".to_string(),
+ )]));
+ assert!(
+ !grant.matches_table(&selected),
+ "a selector travels without setting the flag"
+ );
+
+ let assembled = crate::table::Table::new(
+ table.file_io().clone(),
+ table.identifier().clone(),
+ "/tmp/somewhere-else".to_string(),
+ table.schema().clone(),
+ table.rest_env().cloned(),
+ );
+ assert!(
+ !grant.matches_table(&assembled),
+ "an assembled handle must not replay a grant"
+ );
+
+ let mut branch =
table.copy_with_options(std::collections::HashMap::new());
+ branch.branch_reference = true;
+ assert!(
+ !grant.matches_table(&branch),
+ "a branch view is refused even when its schema id coincides"
+ );
+ }
+
+ fn data_file_for_stats(
+ schema_id: i64,
+ cols: Option<Vec<&str>>,
+ written: Option<Vec<&str>>,
+ ) -> crate::spec::DataFileMeta {
+ crate::spec::DataFileMeta {
+ file_name: "f.parquet".to_string(),
+ file_size: 1,
+ row_count: 1,
+ min_key: Vec::new(),
+ max_key: Vec::new(),
+ key_stats: crate::spec::stats::BinaryTableStats::empty(),
+ value_stats: crate::spec::stats::BinaryTableStats::empty(),
+ min_sequence_number: 0,
+ max_sequence_number: 0,
+ schema_id,
+ level: 0,
+ extra_files: Vec::new(),
+ creation_time: None,
+ delete_row_count: Some(0),
+ embedded_index: None,
+ file_source: None,
+ value_stats_cols: cols.map(|c| c.iter().map(|s|
s.to_string()).collect()),
+ external_path: None,
+ first_row_id: None,
+ write_cols: written.map(|c| c.iter().map(|s|
s.to_string()).collect()),
+ column_max_sequence_numbers: None,
+ }
+ }
+
+ fn plan_of(meta: crate::spec::DataFileMeta) -> crate::table::Plan {
+ crate::table::Plan::new(vec![crate::table::DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(crate::spec::BinaryRowBuilder::new(0).build())
+ .with_bucket(0)
+ .with_bucket_path("p".to_string())
+ .with_total_buckets(1)
+ .with_data_files(vec![meta])
+ .with_raw_convertible(false)
+ .build()
+ .unwrap()])
+ }
+
+ #[tokio::test]
+ async fn test_stats_for_a_dropped_column_are_refused() {
+ let table = query_auth_table();
+ let schemas = table.schema_manager();
+
+ for meta in [
+ data_file_for_stats(table.schema().id(), Some(vec!["id", "gone"]),
None),
+ data_file_for_stats(table.schema().id(), None, Some(vec!["id",
"gone"])),
+ data_file_for_stats(
+ table.schema().id(),
+ Some(vec!["id"]),
+ Some(vec!["id", "gone"]),
+ ),
+ ] {
+ let err = super::reject_unauthorized_stats(&plan_of(meta),
table.schema(), schemas)
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("statistics for 'gone'")),
+ "{err:?}"
+ );
+ }
+
+ assert!(super::reject_unauthorized_stats(
+ &plan_of(data_file_for_stats(
+ table.schema().id(),
+ Some(vec!["id"]),
+ Some(vec!["id"])
+ )),
+ table.schema(),
+ schemas
+ )
+ .await
+ .is_ok());
+ }
+
+ #[tokio::test]
+ async fn test_an_old_schema_whose_column_changed_type_is_refused() {
+ let tmp = tempfile::tempdir().unwrap();
+ let location = tmp.path().display().to_string();
+ let column = |ty| {
+ crate::spec::Schema::builder()
+ .column("id", ty)
+ .option("query-auth.enabled", "true")
+ .build()
+ .unwrap()
+ };
+ let table = crate::table::Table::new(
+ crate::io::FileIOBuilder::new("file").build().unwrap(),
+ crate::catalog::Identifier::new("default", "evolved"),
+ location,
+ crate::spec::TableSchema::new(
+ 0,
+
&column(crate::spec::DataType::Int(crate::spec::IntType::new())),
+ ),
+ None,
+ );
+
+ // Same field id and name, a different type: the server ruled on the
+ // current one, so the older file's stats are not covered.
+ let older = crate::spec::TableSchema::new(
+ 1,
+
&column(crate::spec::DataType::BigInt(crate::spec::BigIntType::new())),
+ );
+ let schemas = table.schema_manager();
+ table
+ .file_io()
+ .new_output(&schemas.schema_path(1))
+ .unwrap()
+ .write(serde_json::to_vec(&older).unwrap().into())
+ .await
+ .unwrap();
+
+ let err = super::reject_unauthorized_stats(
+ &plan_of(data_file_for_stats(1, None, None)),
+ table.schema(),
+ schemas,
+ )
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("statistics for 'id'")),
+ "{err:?}"
+ );
+ }
+
+ #[test]
+ fn test_containment_ignores_comments_and_allows_narrowing() {
+ use crate::spec::{DataField, DataType, IntType, RowType};
+ let int = || DataType::Int(IntType::new());
+ let child = |name: &str, desc: Option<&str>| {
+ let f = DataField::new(1, name.to_string(), int());
+ match desc {
+ Some(d) => f.with_description(Some(d.to_string())),
+ None => f,
+ }
+ };
+ let row = |children: Vec<DataField>|
DataType::Row(RowType::new(children));
+ let wide = row(vec![child("a", None), child("b", None)]);
+
+ // A comment is not a column.
+ assert!(super::contains(
+ &wide,
+ &row(vec![child("a", Some("why")), child("b", None)])
+ ));
+ // Projecting a subset of the children reads nothing extra.
+ assert!(super::contains(&wide, &row(vec![child("a", None)])));
+ // An extra child would.
+ assert!(!super::contains(
+ &wide,
+ &row(vec![
+ child("a", None),
+ child("b", None),
+ child("hidden", None)
+ ])
+ ));
+ // And so would a child under another name.
+ assert!(!super::contains(&wide, &row(vec![child("c", None)])));
+ // Or the same name and type re-added under a new id: older files still
+ // resolve the old id.
+ let readded = DataField::new(9, "a".to_string(), int());
+ assert!(!super::contains(&wide, &row(vec![readded])));
+ }
+
+ #[test]
+ fn test_a_variant_extraction_is_the_one_shape_change_allowed() {
+ use crate::spec::{DataField, DataType, IntType, RowType, VariantType};
+ let int = || DataType::Int(IntType::new());
+ let row = || DataType::Row(RowType::new(vec![DataField::new(0,
"p".into(), int())]));
+ let schema = vec![
+ DataField::new(1, "v".into(),
DataType::Variant(VariantType::new())),
+ DataField::new(2, "n".into(), int()),
+ ];
+ let read = |id, name: &str, ty| vec![DataField::new(id, name.into(),
ty)];
+
+ assert!(super::reject_noncanonical_fields(&read(1, "v", row()),
&schema).is_ok());
+ assert!(super::reject_noncanonical_fields(&read(1, "v", int()),
&schema).is_err());
+ assert!(super::reject_noncanonical_fields(&read(2, "n", row()),
&schema).is_err());
+ }
+
+ #[test]
+ fn test_a_system_column_read_is_refused() {
+ let err = reject_system_columns(["id",
crate::spec::ROW_ID_FIELD_NAME]).unwrap_err();
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("system column '_ROW_ID'")),
+ "{err:?}"
+ );
+ assert!(reject_system_columns(["id", "name"]).is_ok());
+ }
+
+ #[tokio::test]
+ async fn test_time_travelled_or_branch_read_is_refused() {
+ let mut travelled = rest_query_auth_table().await;
+ travelled.time_traveled = true;
+ let err = travelled.authorize_read(true).await.unwrap_err();
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("time-travelled or branch read")),
+ "got {err:?}"
+ );
+
+ let mut branch = rest_query_auth_table().await;
+ branch.branch_reference = true;
+ assert!(branch.authorize_read(true).await.is_err());
+ }
+}
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index d3a308b4..ee54d87a 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -549,11 +549,17 @@ impl<'a> PaimonReadBuilder<'a> {
/// Create a table read for consuming splits (e.g. from a scan plan).
pub fn new_read(&self) -> Result<TableRead<'a>> {
- // Fail closed at read construction so bindings that short-circuit
before
- // `to_arrow` (e.g. an empty-splits fast path) can't bypass the guard.
+ // The declared type needs no grant; only query-auth moved to
`to_arrow`.
let core_options = self.table.schema.core_options();
- core_options.ensure_read_authorized()?;
+
core_options.ensure_type_paimon_served(&self.table.identifier().full_name())?;
core_options.validate_data_file_path_directory()?;
+ // A handle no catalog loaded holds no grant; refused here too, as
bindings
+ // skip `to_arrow` for an empty split list.
+ if core_options.query_auth_enabled() &&
self.table.query_auth_session().is_none() {
+ return Err(super::query_auth::unsupported(
+ "this table handle was assembled rather than loaded",
+ ));
+ }
let read_type = match self.resolve_read_type()? {
None => self.table.schema.fields().to_vec(),
Some(fields) => fields,
@@ -760,6 +766,19 @@ pub(super) fn is_system_projection_field(field_id: i32) ->
bool {
#[cfg(test)]
mod tests {
+ #[tokio::test]
+ async fn
test_new_read_refuses_an_assembled_query_auth_handle_but_not_a_loaded_one() {
+ let assembled = crate::table::query_auth_table();
+ let err = assembled.new_read_builder().new_read().unwrap_err();
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("query-auth.enabled")),
+ "{err:?}"
+ );
+ let loaded = crate::table::rest_query_auth_table().await;
+ assert!(loaded.new_read_builder().new_read().is_ok());
+ }
+
use super::{PaimonReadBuilder, ReadBuilder, ReadBuilderKind};
use crate::table::TableRead;
mod test_utils {
@@ -1003,14 +1022,33 @@ mod tests {
#[test]
fn test_read_fails_closed_when_query_auth_enabled() {
let table = query_auth_table();
- // `new_read` fails closed, so bindings that short-circuit before
`to_arrow` can't bypass.
- let err = table.new_read_builder().new_read().unwrap_err();
+ // An assembled handle is refused at construction; a loaded one at the
read.
+ let err = match table.new_read_builder().new_read() {
+ Err(err) => err,
+ Ok(read) => ungranted_read_error(&read),
+ };
assert!(
matches!(err, crate::Error::Unsupported { ref message } if
message.contains("query-auth.enabled")),
- "building a read for a query-auth.enabled table must fail closed"
+ "reading a query-auth.enabled table without a grant must fail
closed"
);
}
+ fn ungranted_read_error(read: &crate::table::TableRead<'_>) ->
crate::Error {
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path("memory:/t/bucket-0".to_string())
+ .with_total_buckets(1)
+ .with_data_files(Vec::new())
+ .build()
+ .unwrap();
+ match read.to_arrow(&[split]) {
+ Ok(_) => panic!("reading without a grant must fail closed"),
+ Err(err) => err,
+ }
+ }
+
#[test]
fn test_dynamic_option_cannot_disable_query_auth() {
// Copying the table with the option off must not weaken a stored
`true`.
@@ -1018,7 +1056,11 @@ mod tests {
"query-auth.enabled".to_string(),
"false".to_string(),
)]));
- let err = table.new_read_builder().new_read().unwrap_err();
+ // An assembled handle is refused at construction; a loaded one at the
read.
+ let err = match table.new_read_builder().new_read() {
+ Err(err) => err,
+ Ok(read) => ungranted_read_error(&read),
+ };
assert!(
matches!(err, crate::Error::Unsupported { ref message } if
message.contains("query-auth.enabled")),
"a dynamic override must not disable query-auth"
diff --git a/crates/paimon/src/table/rest_env.rs
b/crates/paimon/src/table/rest_env.rs
index f518f0e6..7dfda992 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -19,6 +19,7 @@
use crate::api::rest_api::RESTApi;
use crate::api::rest_error::RestError;
+use crate::api::GetTableResponse;
use crate::catalog::{Identifier, RESTTokenFileIO};
use crate::common::{CatalogOptions, Options};
use crate::error::Error;
@@ -122,6 +123,66 @@ impl RESTEnv {
&self.api
}
+ /// Bracketed by a freshness check: the response names no table, so a
+ /// re-create in between would serve a replacement's grant.
+ pub(crate) async fn table_query_auth(
+ &self,
+ schema_id: i64,
+ fields: &[crate::spec::DataField],
+ select: Option<Vec<String>>,
+ ) -> Result<crate::api::AuthTableQueryResponse> {
+ self.current_table_checked(schema_id, fields).await?;
+ let response = self.api.auth_table_query(&self.identifier,
select).await?;
+ self.current_table_checked(schema_id, fields).await?;
+ Ok(response)
+ }
+
+ /// Refused unless the name still resolves to the loaded table, a missing
+ /// identity included. Asserts nothing on its own.
+ pub(crate) async fn current_table_checked(
+ &self,
+ schema_id: i64,
+ fields: &[crate::spec::DataField],
+ ) -> Result<GetTableResponse> {
+ let response = self.api.get_table(&self.identifier).await?;
+ let name = self.identifier.full_name();
+ let same = |what: &str, loaded: String, now: Option<String>| match now
{
+ Some(now) if now == loaded => Ok(()),
+ now => Err(crate::Error::DataInvalid {
+ message: format!(
+ "table '{name}' now resolves to {what} {}, not the
{loaded} this handle was \
+ loaded with; re-load the table before reading it",
+ now.as_deref().unwrap_or("nothing the server reports")
+ ),
+ source: None,
+ }),
+ };
+ same("uuid", self.uuid.clone(), response.id.clone())?;
+ same(
+ "schema",
+ schema_id.to_string(),
+ response.schema_id.map(|id| id.to_string()),
+ )?;
+ // An id is not the schema: the columns the server rules on are
compared too.
+ let key =
+ |f: &crate::spec::DataField| (f.id(), f.name().to_string(),
f.data_type().clone());
+ let served: Vec<_> = response
+ .schema
+ .as_ref()
+ .map(|schema| schema.fields().iter().map(key).collect())
+ .unwrap_or_default();
+ if served != fields.iter().map(key).collect::<Vec<_>>() {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "table '{name}' serves other columns than this handle
carries under schema \
+ {schema_id}; re-load the table before reading it"
+ ),
+ source: None,
+ });
+ }
+ Ok(response)
+ }
+
/// Get the table identifier.
pub fn identifier(&self) -> &Identifier {
&self.identifier
@@ -177,8 +238,6 @@ impl RESTEnv {
.map_err(|e| map_rest_error_for_table(e, identifier))
}
- /// Build a Table from an already-fetched response, so routing can
- /// inspect the declared type first.
pub(crate) async fn build_table(
identifier: &Identifier,
response: crate::api::GetTableResponse,
@@ -276,9 +335,10 @@ impl RESTEnv {
Some(rest_env),
);
+ // Minted after the schema-replacing copy, which drops any session.
let mut table =
table.copy_with_resolved_schema(table.schema().clone(), &branch)?;
table.branch_reference = branch_reference;
- Ok(table)
+ Ok(table.with_query_auth_session())
}
pub(crate) async fn build_object_table(
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index f96a4f47..ab501eeb 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -20,6 +20,7 @@
//! Reference:
[org.apache.paimon.table.source](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/table/source/).
use crate::spec::{BinaryRow, DataFileMeta, DataFileMetaRowLayout};
+use crate::table::query_auth::QueryAuthGrant;
use crate::table::stats_filter::group_by_overlapping_row_id;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
@@ -507,6 +508,14 @@ pub struct DataSplit {
raw_convertible: bool,
#[serde(default)]
is_streaming: bool,
+ /// Mirrors Java `QueryAuthSplit`, but is dropped by serialization, so a
+ /// plan must be read where it was made.
+ #[serde(skip)]
+ query_auth_grant: Option<Arc<QueryAuthGrant>>,
+ /// That this split came from a `query-auth.enabled` table. Unlike the
grant
+ /// it survives serialization, so a round-tripped split fails closed.
+ #[serde(default)]
+ query_auth_required: bool,
}
impl DataSplit {
@@ -515,6 +524,22 @@ impl DataSplit {
self.is_streaming
}
+ /// Needs authorization even without a grant, so a plan that produced none
+ /// still refuses at the read.
+ pub(crate) fn planned(mut self, grant: Option<Arc<QueryAuthGrant>>) ->
Self {
+ self.query_auth_required = true;
+ self.query_auth_grant = grant;
+ self
+ }
+
+ pub(crate) fn query_auth_required(&self) -> bool {
+ self.query_auth_required
+ }
+
+ pub(crate) fn query_auth_grant(&self) -> Option<&Arc<QueryAuthGrant>> {
+ self.query_auth_grant.as_ref()
+ }
+
pub fn snapshot_id(&self) -> i64 {
self.snapshot_id
}
@@ -709,10 +734,23 @@ 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.
+ fn ensure_serializable_without_grant(&self) -> crate::Result<()> {
+ if self.query_auth_required {
+ return Err(crate::table::query_auth::unsupported(
+ "a split of such a table cannot be serialized to the
cross-language \
+ format, which has no field to carry the authorization with
it",
+ ));
+ }
+ Ok(())
+ }
+
/// Serialize the DataSplit fields to Java `DataSplit#serialize` (version
9) binary.
/// Byte-compatible with `compatibility/datasplit-v9`. Row ranges are not
part of the
/// format; `serialize_split_v1` wraps a row-range split as an
`IndexedSplit` instead.
pub fn serialize(&self) -> crate::Result<Vec<u8>> {
+ self.ensure_serializable_without_grant()?;
let mut out = Vec::new();
out.extend_from_slice(&SPLIT_MAGIC.to_be_bytes());
out.extend_from_slice(&SPLIT_VERSION.to_be_bytes());
@@ -868,6 +906,7 @@ impl DataSplit {
/// `IndexedSplit` (type 3) wrapping the DataSplit body plus the ranges.
Byte-compatible with
/// `compatibility/split-v1-data` / `split-v1-indexed`.
pub fn serialize_split_v1(&self) -> crate::Result<Vec<u8>> {
+ self.ensure_serializable_without_grant()?;
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());
@@ -1302,6 +1341,8 @@ impl DataSplitBuilder {
}
}
Ok(DataSplit {
+ query_auth_grant: None,
+ query_auth_required: false,
snapshot_id: self.snapshot_id,
partition: Arc::new(partition),
bucket: self.bucket,
@@ -1357,6 +1398,17 @@ impl Plan {
&self.splits
}
+ /// Stamps the grant every split of this plan was authorized under.
+ pub(crate) fn planned(mut self, grant: Option<Arc<QueryAuthGrant>>) ->
Self {
+ if grant.is_some() {
+ self.splits = std::mem::take(&mut self.splits)
+ .into_iter()
+ .map(|split| split.planned(grant.clone()))
+ .collect();
+ }
+ self
+ }
+
/// Sum of data-file bytes referenced by this plan.
///
/// Negative file sizes are treated as unknown and do not contribute. The
@@ -2030,6 +2082,31 @@ mod tests {
}
// Same hardening for the IndexedSplit row-ranges count in the SPLIT_V1
frame.
+ #[test]
+ fn test_a_marked_split_refuses_the_cross_language_formats() {
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(crate::spec::BinaryRowBuilder::new(0).build())
+ .with_bucket(0)
+ .with_bucket_path("p".to_string())
+ .with_total_buckets(1)
+ .with_data_files(vec![])
+ .with_raw_convertible(false)
+ .build()
+ .unwrap()
+ .planned(None);
+ for bytes in [split.serialize(), split.serialize_split_v1()] {
+ let Err(err) = bytes else {
+ panic!("the marker has nowhere to go in these formats")
+ };
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("query-auth.enabled")),
+ "{err:?}"
+ );
+ }
+ }
+
#[test]
fn deserialize_split_v1_rejects_huge_ranges_count_without_aborting() {
let split = DataSplitBuilder::new()
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index e56b9477..eef8c266 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -354,6 +354,8 @@ impl TableCommit {
});
}
// A commit validates against the existing snapshot.
+ // A refusal here must not clean up: a retry with an identifier that
+ // already committed names files a snapshot references.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
self.table.ensure_not_branch_reference_for_write()?;
reject_compact_increment(&commit_messages)?;
@@ -6340,6 +6342,54 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_a_refused_retry_keeps_the_files_its_identifier_committed() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_refused_retry_keeps_committed_files";
+ setup_dirs(&file_io, table_path).await;
+ let index_path = format!("{table_path}/index/bucket-index");
+ file_io
+ .mkdirs(&format!("{table_path}/index/"))
+ .await
+ .unwrap();
+ file_io
+ .new_output(&index_path)
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"index"))
+ .await
+ .unwrap();
+ let mut message = CommitMessage::new(vec![], 0, vec![]);
+ message.new_index_files = vec![IndexFileMeta {
+ index_type: "HASH".to_string(),
+ file_name: "bucket-index".to_string(),
+ file_size: 5,
+ row_count: 1,
+ deletion_vectors_ranges: None,
+ external_path: None,
+ global_index_meta: None,
+ }];
+ setup_commit(&file_io, table_path)
+ .commit_with_identifier(vec![message.clone()], 7)
+ .await
+ .unwrap();
+
+ // The option arrives between the commit and its retry.
+ let guarded = test_table_with_options(
+ &file_io,
+ table_path,
+ HashMap::from([("query-auth.enabled".to_string(),
"true".to_string())]),
+ );
+ let retry = TableCommit::new(guarded, "test-user".to_string());
+ assert!(retry
+ .filter_and_commit_with_identifier(vec![message], 7)
+ .await
+ .is_err());
+ assert!(
+ file_io.exists(&index_path).await.unwrap(),
+ "a refused retry must not delete files the first commit's snapshot
references"
+ );
+ }
+
#[tokio::test]
async fn test_abort_deletes_index_files_from_the_data_file_directory() {
// With `index-file-in-data-file-dir`, a new index file is written
beside the
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index c7516df4..ee0d3a0f 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -224,6 +224,9 @@ impl<'a> TableRead<'a> {
}
/// Returns an [`ArrowRecordBatchStream`].
+ ///
+ /// A `query-auth.enabled` table reads only splits from this handle's plan;
+ /// re-plan after a permission change.
pub fn to_arrow(&self, data_splits: &[DataSplit]) ->
crate::Result<ArrowRecordBatchStream> {
match &self.0 {
TableReadKind::Paimon(read) => read.to_arrow(data_splits),
@@ -240,7 +243,7 @@ impl<'a> TableRead<'a> {
&self,
data_splits: &[DataSplit],
) -> crate::Result<ArrowRecordBatchStream> {
- self.ensure_query_auth_allowed()?;
+ // Decided from the splits by the `to_arrow` each branch ends in.
match &self.0 {
TableReadKind::Paimon(read) =>
read.to_arrow_with_row_kind(data_splits),
TableReadKind::Format(read) => {
@@ -258,7 +261,7 @@ impl<'a> TableRead<'a> {
&self,
plan: &IncrementalPlan,
) -> crate::Result<ArrowRecordBatchStream> {
- self.ensure_query_auth_allowed()?;
+ self.ensure_query_auth_allowed(plan)?;
plan.validate()?;
match &self.0 {
TableReadKind::Paimon(read) => read.to_incremental_arrow(plan),
@@ -278,7 +281,7 @@ impl<'a> TableRead<'a> {
&self,
plan: &IncrementalPlan,
) -> crate::Result<ArrowRecordBatchStream> {
- self.ensure_query_auth_allowed()?;
+ self.ensure_query_auth_allowed(plan)?;
plan.validate()?;
match &self.0 {
TableReadKind::Paimon(read) => read.to_audit_log_arrow(plan),
@@ -288,8 +291,40 @@ impl<'a> TableRead<'a> {
}
}
- fn ensure_query_auth_allowed(&self) -> crate::Result<()> {
-
CoreOptions::new(self.table().schema().options()).ensure_read_authorized()
+ /// The split's marker carries the plan's decision.
+ fn ensure_query_auth_allowed(&self, plan: &IncrementalPlan) ->
crate::Result<()> {
+ let core_options = CoreOptions::new(self.table().schema().options());
+
core_options.ensure_type_paimon_served(&self.table().identifier().full_name())?;
+ // Diff pairs too, which `data_splits()` leaves out.
+ let marked = plan.splits().iter().any(|split| match split {
+ IncrementalSplit::Data(split) => split.query_auth_required(),
+ IncrementalSplit::DiffPair { before, after } => before
+ .iter()
+ .chain(after)
+ .any(DataSplit::query_auth_required),
+ });
+ if core_options.query_auth_enabled() || marked {
+ return Err(super::query_auth::unsupported(
+ "an incremental read cannot apply a row filter or column
masking",
+ ));
+ }
+ Ok(())
+ }
+}
+
+/// Every leaf's column name, system columns included.
+fn collect_leaf_column_names(predicate: &Predicate, out: &mut
std::collections::HashSet<String>) {
+ match predicate {
+ Predicate::Leaf { column, .. } => {
+ out.insert(column.clone());
+ }
+ Predicate::And(children) | Predicate::Or(children) => {
+ children
+ .iter()
+ .for_each(|child| collect_leaf_column_names(child, out));
+ }
+ Predicate::Not(inner) => collect_leaf_column_names(inner, out),
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => {}
}
}
@@ -472,6 +507,8 @@ impl<'a> PaimonTableRead<'a> {
&self,
data_splits: &[DataSplit],
) -> crate::Result<ArrowRecordBatchStream> {
+ // Streaming primary-key splits are read raw below, so decide here for
all.
+ self.ensure_authorized_by_splits(&self.table.schema.core_options(),
data_splits)?;
let schema = audit_schema_for_read_type(&self.read_type, false)?;
let (streaming, materialized): (Vec<_>, Vec<_>) = data_splits
.iter()
@@ -852,12 +889,74 @@ impl<'a> PaimonTableRead<'a> {
reader.read(splits)
}
+ /// Reads only splits carrying a grant that the server imposed nothing;
+ /// nothing is fetched here, so a split without one fails closed.
+ fn ensure_authorized_by_splits(
+ &self,
+ core_options: &CoreOptions,
+ data_splits: &[DataSplit],
+ ) -> crate::Result<()> {
+ // Unconditional: unrelated to query-auth.
+
core_options.ensure_type_paimon_served(&self.table.identifier().full_name())?;
+ // Decided at plan time, as in Java: a split predating the option, or
built
+ // by hand, carries neither flag and is read on the caller's word.
+ let required = core_options.query_auth_enabled()
+ || data_splits.iter().any(|s| s.query_auth_required());
+ if !required {
+ return Ok(());
+ }
+ // Only the catalog mints a session, so a handle without one holds no
grant.
+ if self.table.query_auth_session().is_none() {
+ return Err(super::query_auth::unsupported(
+ "this table handle was assembled rather than loaded",
+ ));
+ }
+ // The read's own scope: a caller can plan clean, then read
differently.
+ let mut filter_columns = std::collections::HashSet::new();
+ for predicate in &self.data_predicates {
+ collect_leaf_column_names(predicate, &mut filter_columns);
+ }
+ super::query_auth::reject_system_columns(
+ self.read_type
+ .iter()
+ .map(|f| f.name())
+ .chain(filter_columns.iter().map(String::as_str)),
+ )?;
+ // By id and name: older files resolve by id, so a dropped field passed
+ // to `with_read_type` would read an uncovered column.
+ super::query_auth::reject_noncanonical_fields(
+ &self.read_type,
+ self.table.schema().fields(),
+ )?;
+ // Per split, as Java binds one `QueryAuthSplit` each.
+ for split in data_splits {
+ let Some(grant) = split.query_auth_grant() else {
+ return Err(super::query_auth::unsupported(
+ "the split carries no authorization; it was built
directly, or serialized, \
+ which drops the grant — re-plan the scan",
+ ));
+ };
+ if !grant.matches_table(self.table) {
+ return Err(super::query_auth::unsupported(
+ "the grant was issued for a different table, schema or
session; re-plan the \
+ scan",
+ ));
+ }
+ if !grant.is_unrestricted() {
+ return Err(super::query_auth::unsupported(
+ "this client cannot apply a row filter or column masking,
so it refuses \
+ rather than return unfiltered rows",
+ ));
+ }
+ }
+ Ok(())
+ }
+
/// Returns an [`ArrowRecordBatchStream`].
pub fn to_arrow(&self, data_splits: &[DataSplit]) ->
crate::Result<ArrowRecordBatchStream> {
let has_primary_keys = !self.table.schema.primary_keys().is_empty();
let core_options = self.table.schema.core_options();
- // Fail closed for a direct `TableRead` (bypassing
`ReadBuilder::new_read`).
- core_options.ensure_read_authorized()?;
+ self.ensure_authorized_by_splits(&core_options, data_splits)?;
let merge_engine = core_options.merge_engine()?;
// Route supported PK merge engines through the split-aware reader.
@@ -2103,15 +2202,364 @@ mod tests {
));
}
+ #[test]
+ fn test_an_audit_log_read_refuses_a_query_auth_table() {
+ let table = query_auth_table();
+ let read = TableRead::new(&table, table.schema.fields().to_vec(),
Vec::new());
+ let audit =
+ AuditLogRead::new(read).expect("the decision is per split, not at
construction");
+ // An empty split list reads as empty; a split without a grant does
not.
+ let Err(err) = audit.to_arrow(&[split_with_grant(None)]) else {
+ panic!("an audit read must refuse a query-auth.enabled table")
+ };
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("query-auth.enabled")),
+ "{err:?}"
+ );
+ }
+
+ #[test]
+ fn test_an_audit_log_read_checks_marked_splits_on_a_primary_key_table() {
+ // The primary-key path builds its readers itself, so it must not skip
+ // the split-carried decision `PaimonTableRead::to_arrow` makes.
+ let table = file_index_table("memory:/table_read_audit_marked_split",
None, true);
+ let read = TableRead::new(&table, table.schema().fields().to_vec(),
Vec::new());
+ let audit = AuditLogRead::new(read).unwrap();
+ let Err(err) = audit.to_arrow(&[split_with_grant(None)]) else {
+ panic!("a marked split without a grant must be refused")
+ };
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("query-auth.enabled")),
+ "{err:?}"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_an_assembled_query_auth_handle_refuses_even_an_empty_read() {
+ // A handle no catalog minted a session for cannot be authorized, so an
+ // empty split list is not an empty result but a refusal.
+ let table = query_auth_table();
+ let read = TableRead::new(&table, table.schema.fields().to_vec(),
Vec::new());
+ let Err(err) = read.to_arrow(&[]) else {
+ panic!("an assembled query-auth handle must refuse")
+ };
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("query-auth.enabled")),
+ "{err:?}"
+ );
+ // A loaded handle with an empty plan still reads as empty.
+ let loaded = crate::table::rest_query_auth_table().await;
+ let read = TableRead::new(&loaded, loaded.schema.fields().to_vec(),
Vec::new());
+ assert!(read.to_arrow(&[]).is_ok());
+ }
+
+ #[test]
+ fn test_a_marked_split_refuses_the_row_kind_read() {
+ // A streaming split of a primary-key table is the path that reads raw.
+ let paimon = file_index_table("memory:/table_read_row_kind_marked",
None, true);
+ let read = TableRead::new(&paimon, paimon.schema().fields().to_vec(),
Vec::new());
+ let marked = crate::table::DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(crate::spec::BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path("memory:/t/bucket-0".to_string())
+ .with_total_buckets(1)
+ .with_data_files(Vec::new())
+ .with_streaming(true)
+ .build()
+ .unwrap()
+ .planned(None);
+ let Err(err) = read.to_arrow_with_row_kind(&[marked]) else {
+ panic!("a marked split must refuse a row-kind read")
+ };
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("query-auth.enabled")),
+ "{err:?}"
+ );
+ }
+
+ fn stale_handle(name: &str, options: &[(&str, &str)]) -> Table {
+ let mut builder = crate::spec::Schema::builder().column(
+ "id",
+ crate::spec::DataType::Int(crate::spec::IntType::new()),
+ );
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ Table::new(
+ FileIOBuilder::new("file").build().unwrap(),
+ Identifier::new("default", name),
+ format!("/tmp/test-{name}"),
+ crate::spec::TableSchema::new(0, &builder.build().unwrap()),
+ None,
+ )
+ }
+
+ #[test]
+ fn test_a_marked_split_refuses_the_format_and_incremental_reads() {
+ let stamped = split_with_grant(None);
+
+ let format = stale_handle(
+ "fmt",
+ &[("type", "format-table"), ("file.format", "parquet")],
+ );
+ let read =
+ TableRead::new_format(&format, format.schema().fields().to_vec(),
Vec::new(), None);
+ let Err(err) = read.to_arrow(std::slice::from_ref(&stamped)) else {
+ panic!("a marked split must refuse a format read")
+ };
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("query-auth.enabled")),
+ "{err:?}"
+ );
+
+ let paimon = stale_handle("inc", &[]);
+ let read = TableRead::new(&paimon, paimon.schema().fields().to_vec(),
Vec::new());
+ let plan = IncrementalPlan::new(
+ IncrementalScanMode::Delta,
+ vec![IncrementalSplit::Data(stamped)],
+ );
+ let Err(err) = read.to_incremental_arrow(&plan) else {
+ panic!("a marked split must refuse an incremental read")
+ };
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("query-auth.enabled")),
+ "{err:?}"
+ );
+ }
+
+ fn split_with_grant(
+ grant: Option<crate::table::query_auth::QueryAuthGrant>,
+ ) -> crate::table::DataSplit {
+ crate::table::DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(crate::spec::BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path("memory:/t/bucket-0".to_string())
+ .with_total_buckets(1)
+ .with_data_files(Vec::new())
+ .build()
+ .unwrap()
+ .planned(grant.map(std::sync::Arc::new))
+ }
+
+ fn grant_for(table: &Table, restricted: bool) ->
crate::table::query_auth::QueryAuthGrant {
+ crate::table::query_auth::QueryAuthGrant::new(
+ crate::api::AuthTableQueryResponse {
+ filter: restricted.then(|| vec!["{}".to_string()]),
+ column_masking: None,
+ },
+ table
+ .query_auth_session()
+ .expect("a catalog-loaded table has a session"),
+ )
+ }
+
+ #[tokio::test]
+ async fn test_one_unrestricted_grant_does_not_cover_the_other_splits() {
+ let table = crate::table::rest_query_auth_table().await;
+ let read = TableRead::new(&table, table.schema.fields().to_vec(),
Vec::new());
+ let allowed = split_with_grant(Some(grant_for(&table, false)));
+ for other in [
+ split_with_grant(Some(grant_for(&table, true))),
+ split_with_grant(None),
+ ] {
+ assert!(
+ read.to_arrow(&[allowed.clone(), other]).is_err(),
+ "every split must be authorized on its own"
+ );
+ }
+ }
+
+ #[test]
+ fn test_engine_served_table_is_refused_at_the_read_boundary() {
+ let schema = crate::spec::Schema::builder()
+ .column(
+ "id",
+ crate::spec::DataType::Int(crate::spec::IntType::new()),
+ )
+ .option("type", "iceberg-table")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ crate::io::FileIOBuilder::new("file").build().unwrap(),
+ crate::catalog::Identifier::new("default", "ice_t"),
+ "/tmp/test-engine-served-read".to_string(),
+ crate::spec::TableSchema::new(0, &schema),
+ None,
+ );
+ let read = TableRead::new(&table, table.schema.fields().to_vec(),
Vec::new());
+ let err = match read.to_arrow(&[split_with_grant(None)]) {
+ Ok(_) => panic!("an engine-served table must not be read as
Paimon"),
+ Err(err) => err,
+ };
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("cannot be served as a Paimon table")),
+ "got {err:?}"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_a_row_id_filter_configured_on_the_read_is_refused() {
+ let table = crate::table::rest_query_auth_table().await;
+ let row_id = Predicate::Leaf {
+ index: 0,
+ column: crate::spec::ROW_ID_FIELD_NAME.to_string(),
+ data_type:
crate::spec::DataType::BigInt(crate::spec::BigIntType::new()),
+ op: crate::spec::PredicateOperator::GtEq,
+ literals: vec![crate::spec::Datum::Long(1)],
+ };
+ let read = TableRead::new(&table, table.schema.fields().to_vec(),
vec![row_id]);
+ let split = split_with_grant(Some(grant_for(&table, false)));
+ let err = match read.to_arrow(&[split]) {
+ Ok(_) => panic!("a system-column filter must be refused"),
+ Err(err) => err,
+ };
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("system column '_ROW_ID'")),
+ "got {err:?}"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_a_field_outside_the_current_schema_is_refused() {
+ let table = crate::table::rest_query_auth_table().await;
+ let dropped = crate::spec::DataField::new(
+ 999,
+ "dropped".to_string(),
+ crate::spec::DataType::Int(crate::spec::IntType::new()),
+ );
+ let real = table.schema().fields()[0].clone();
+ let reshaped = crate::spec::DataField::new(
+ real.id(),
+ real.name().to_string(),
+ crate::spec::DataType::Row(crate::spec::RowType::new(vec![
+ crate::spec::DataField::new(
+ 1,
+ "hidden".to_string(),
+ crate::spec::DataType::Int(crate::spec::IntType::new()),
+ ),
+ ])),
+ );
+
+ for field in [dropped, reshaped] {
+ let read = TableRead::new(&table, vec![field], Vec::new());
+ let split = split_with_grant(Some(grant_for(&table, false)));
+ let err = match read.to_arrow(&[split]) {
+ Ok(_) => panic!("a field outside the current schema must be
refused"),
+ Err(err) => err,
+ };
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("not a column of the current schema")),
+ "got {err:?}"
+ );
+ }
+ }
+
+ #[test]
+ fn test_a_serialized_split_still_demands_authorization() {
+ let stamped = split_with_grant(None);
+ let bytes = serde_json::to_vec(&stamped).unwrap();
+ let restored: crate::table::DataSplit =
serde_json::from_slice(&bytes).unwrap();
+ assert!(
+ restored.query_auth_grant().is_none(),
+ "the grant is dropped"
+ );
+ assert!(
+ restored.query_auth_required(),
+ "but the demand for authorization survives"
+ );
+
+ let stale = Table::new(
+ crate::io::FileIOBuilder::new("file").build().unwrap(),
+ crate::catalog::Identifier::new("default", "stale"),
+ "/tmp/test-stale-handle".to_string(),
+ crate::spec::TableSchema::new(
+ 0,
+ &crate::spec::Schema::builder()
+ .column(
+ "id",
+
crate::spec::DataType::Int(crate::spec::IntType::new()),
+ )
+ .build()
+ .unwrap(),
+ ),
+ None,
+ );
+ assert!(!stale.schema().core_options().query_auth_enabled());
+ let read = TableRead::new(&stale, stale.schema().fields().to_vec(),
Vec::new());
+ assert!(
+ read.to_arrow(&[restored]).is_err(),
+ "a round-tripped split must fail closed even on a handle that
predates the option"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_a_grant_from_another_handle_refuses_the_read() {
+ let table = crate::table::rest_query_auth_table().await;
+ let other = crate::table::rest_query_auth_table().await;
+ let read = TableRead::new(&table, table.schema.fields().to_vec(),
Vec::new());
+ let err = match
read.to_arrow(&[split_with_grant(Some(grant_for(&other, false)))]) {
+ Ok(_) => panic!("a grant obtained elsewhere must not authorize
this read"),
+ Err(err) => err,
+ };
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("different table, schema or session")),
+ "got {err:?}"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_restricted_grant_on_a_split_refuses_the_read() {
+ let table = crate::table::rest_query_auth_table().await;
+ let read = TableRead::new(&table, table.schema.fields().to_vec(),
Vec::new());
+ let split = split_with_grant(Some(grant_for(&table, true)));
+ assert!(
+ matches!(
+ read.to_arrow(&[split]),
+ Err(crate::Error::Unsupported { ref message }) if
message.contains("query-auth.enabled")
+ ),
+ "a row filter this client cannot apply must refuse the read"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_unrestricted_grant_on_a_split_allows_the_read() {
+ let table = crate::table::rest_query_auth_table().await;
+ let read = TableRead::new(&table, table.schema.fields().to_vec(),
Vec::new());
+ let split = split_with_grant(Some(grant_for(&table, false)));
+ assert!(
+ read.to_arrow(&[split]).is_ok(),
+ "an unrestricted grant must let the read through"
+ );
+ }
+
#[test]
fn test_direct_table_read_fails_closed_when_query_auth_enabled() {
let table = query_auth_table();
- // Bypass `ReadBuilder` by constructing `TableRead` directly; the
`to_arrow` guard
- // still fails closed.
let read = TableRead::new(&table, table.schema.fields().to_vec(),
Vec::new());
+ let split = crate::table::DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(crate::spec::BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path("memory:/t/bucket-0".to_string())
+ .with_total_buckets(1)
+ .with_data_files(Vec::new())
+ .build()
+ .unwrap();
assert!(
matches!(
- read.to_arrow(&[]),
+ read.to_arrow(&[split]),
Err(crate::Error::Unsupported { ref message }) if
message.contains("query-auth.enabled")
),
"directly-constructed read of a query-auth.enabled table must fail
closed"
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index ceb4bc85..6205cd11 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -1268,15 +1268,16 @@ impl<'a> TableScan<'a> {
}
pub async fn plan(&self) -> crate::Result<Plan> {
+ // Boxed: engines poll this under deep operator stacks.
match &self.0 {
- TableScanKind::Paimon(scan) => scan.plan().await,
+ TableScanKind::Paimon(scan) => Box::pin(scan.plan()).await,
TableScanKind::Format(scan) => scan.plan().await,
}
}
pub async fn plan_with_trace(&self) -> crate::Result<(Plan, ScanTrace)> {
match &self.0 {
- TableScanKind::Paimon(scan) => scan.plan_with_trace().await,
+ TableScanKind::Paimon(scan) =>
Box::pin(scan.plan_with_trace()).await,
TableScanKind::Format(scan) => scan.plan_with_trace().await,
}
}
@@ -1499,45 +1500,111 @@ impl<'a> PaimonTableScan<'a> {
/// for `scan.version`; the strict selectors mirror Java's typed
/// `scan.snapshot-id` / `scan.tag-name` handling.
pub async fn plan(&self) -> crate::Result<Plan> {
- self.validate_read_options()?;
+ let grant = self.authorize_query().await?;
self.validate_shard_strategy()?;
let data_evolution_read_field_ids = self.projected_read_field_ids()?;
- let snapshot = match
super::time_travel::resolve_snapshot(self.table).await? {
- Some(snapshot) => snapshot,
- None => return Ok(Plan::new(Vec::new())),
+ let plan = match
super::time_travel::resolve_snapshot(self.table).await? {
+ Some(snapshot) => {
+ self.plan_snapshot(snapshot,
data_evolution_read_field_ids.as_ref(), None)
+ .await?
+ }
+ None => Plan::new(Vec::new()),
};
- self.plan_snapshot(snapshot, data_evolution_read_field_ids.as_ref(),
None)
- .await
+ self.check_planned_files(&plan, grant.is_some()).await?;
+ Ok(plan.planned(grant))
}
/// Plan the full scan and return metadata-pruning trace counters.
pub async fn plan_with_trace(&self) -> crate::Result<(Plan, ScanTrace)> {
- self.validate_read_options()?;
+ let grant = self.authorize_query().await?;
self.validate_shard_strategy()?;
let mut trace = ScanTrace {
limit: self.limit,
..Default::default()
};
let data_evolution_read_field_ids = self.projected_read_field_ids()?;
- let snapshot = match
super::time_travel::resolve_snapshot(self.table).await? {
- Some(snapshot) => snapshot,
- None => return Ok((Plan::new(Vec::new()), trace)),
+ let plan = match
super::time_travel::resolve_snapshot(self.table).await? {
+ Some(snapshot) => {
+ trace.snapshot_id = Some(snapshot.id());
+ let plan = self
+ .plan_snapshot(
+ snapshot,
+ data_evolution_read_field_ids.as_ref(),
+ Some(&mut trace),
+ )
+ .await?;
+ trace.planned_data_file_bytes = plan.planned_data_file_bytes();
+ plan
+ }
+ None => Plan::new(Vec::new()),
};
- trace.snapshot_id = Some(snapshot.id());
- let plan = self
- .plan_snapshot(
- snapshot,
- data_evolution_read_field_ids.as_ref(),
- Some(&mut trace),
- )
- .await?;
- trace.planned_data_file_bytes = plan.planned_data_file_bytes();
- Ok((plan, trace))
+ self.check_planned_files(&plan, grant.is_some()).await?;
+ Ok((plan.planned(grant), trace))
+ }
+
+ /// The grant predates the manifest read, so the table can have been
+ /// re-created in between. Also refuses stats the schema no longer covers.
+ async fn check_planned_files(&self, plan: &Plan, query_auth: bool) ->
crate::Result<()> {
+ if !query_auth {
+ return Ok(());
+ }
+ if let Some(rest_env) = self.table.rest_env() {
+ rest_env
+ .current_table_checked(self.table.schema().id(),
self.table.schema().fields())
+ .await?;
+ }
+ super::query_auth::reject_unauthorized_stats(
+ plan,
+ self.table.schema(),
+ self.table.schema_manager(),
+ )
+ .await
+ }
+
+ /// Authorize this scan and return the grant for the caller to stamp.
+ async fn authorize_query(
+ &self,
+ ) -> crate::Result<Option<Arc<super::query_auth::QueryAuthGrant>>> {
+ let core_options = CoreOptions::new(self.table.schema().options());
+ // Unconditional: unrelated to query-auth.
+
core_options.ensure_type_paimon_served(&self.table.identifier().full_name())?;
+ core_options.validate_data_file_path_directory()?;
+
+ // File paths and stats, not table columns: the endpoint cannot rule
on them.
+ let query_auth =
CoreOptions::new(self.table.schema().options()).query_auth_enabled();
+ if self.scan_all_files {
+ return if query_auth {
+ Err(super::query_auth::unsupported(
+ "`$files` and friends are file paths and stats, not table
columns, so the \
+ auth endpoint can never rule on them",
+ ))
+ } else {
+ Ok(None)
+ };
+ }
+ // A predicate or a row-range slice reads `_ROW_ID` unprojected.
+ let touches_row_id = self.row_ranges.is_some()
+ || self
+ .data_predicates
+ .iter()
+ .any(super::row_id_predicate::references_row_id);
+ if query_auth && touches_row_id {
+ super::query_auth::reject_system_columns([ROW_ID_FIELD_NAME])?;
+ }
+
+ let grant = self.table.authorize_read(query_auth).await?;
+ // A plan already answers COUNT/MIN/MAX from row counts and bounds.
+ if grant.as_ref().is_some_and(|g| !g.is_unrestricted()) {
+ return Err(super::query_auth::unsupported(
+ "a plan already carries file paths, row counts and column
bounds that a row \
+ filter or column masking must not expose",
+ ));
+ }
+ Ok(grant)
}
- /// Fail closed for a `query-auth.enabled` table: scan planning — including
- /// `with_scan_all_files`, which read-facing system tables like `files`
use —
- /// exposes file paths, row counts, and stats the client can't authorize.
+ /// Fail closed on planning paths that do not authorize,
`with_scan_all_files`
+ /// included.
fn validate_read_options(&self) -> crate::Result<()> {
let options = CoreOptions::new(self.table.schema().options());
options.ensure_read_authorized()?;
@@ -2777,6 +2844,36 @@ mod tests {
use chrono::{DateTime, Utc};
use std::collections::{HashMap, HashSet};
+ #[tokio::test]
+ async fn test_engine_served_table_is_refused_at_plan() {
+ let schema = crate::spec::Schema::builder()
+ .column(
+ "id",
+ crate::spec::DataType::Int(crate::spec::IntType::new()),
+ )
+ .option("type", "iceberg-table")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ crate::io::FileIOBuilder::new("file").build().unwrap(),
+ crate::catalog::Identifier::new("default", "ice_t"),
+ "/tmp/test-engine-served".to_string(),
+ crate::spec::TableSchema::new(0, &schema),
+ None,
+ );
+ let err = table
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .expect_err("an engine-served table must not plan as Paimon");
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("cannot be served as a Paimon table")),
+ "got {err:?}"
+ );
+ }
+
/// Helper to build a DataFileMeta with data evolution fields.
fn make_evo_file(
name: &str,
diff --git a/crates/paimon/tests/mock_server.rs
b/crates/paimon/tests/mock_server.rs
index 893c61ca..68dadf34 100644
--- a/crates/paimon/tests/mock_server.rs
+++ b/crates/paimon/tests/mock_server.rs
@@ -34,14 +34,15 @@ use std::sync::{Arc, Mutex};
use tokio::task::JoinHandle;
use paimon::api::{
- AlterDatabaseRequest, AlterTableRequest, AuditRESTResponse, ConfigResponse,
- CreateFunctionRequest, CreatePartitionsRequest, CreateTagRequest,
CreateViewRequest,
- DataPolicy, DropPartitionsRequest, DropPolicyRequest, ErrorResponse,
GetDatabaseResponse,
- GetFunctionResponse, GetTableResponse, GetTagResponse, GetViewResponse,
ListDatabasesResponse,
- ListFunctionsResponse, ListPartitionsByFilterRequest,
ListPartitionsByNamesRequest,
- ListPartitionsResponse, ListPermissionsResponse, ListPoliciesResponse,
ListTablesResponse,
- ListViewsResponse, PermissionAssignment, PermissionResource,
PolicyRequest, PolicyType,
- RenameTableRequest, ResourcePaths, ResourceType, RevokePermissionRequest,
+ AlterDatabaseRequest, AlterTableRequest, AuditRESTResponse,
AuthTableQueryResponse,
+ ConfigResponse, CreateFunctionRequest, CreatePartitionsRequest,
CreateTagRequest,
+ CreateViewRequest, DataPolicy, DropPartitionsRequest, DropPolicyRequest,
ErrorResponse,
+ GetDatabaseResponse, GetFunctionResponse, GetTableResponse,
GetTagResponse, GetViewResponse,
+ ListDatabasesResponse, ListFunctionsResponse,
ListPartitionsByFilterRequest,
+ ListPartitionsByNamesRequest, ListPartitionsResponse,
ListPermissionsResponse,
+ ListPoliciesResponse, ListTablesResponse, ListViewsResponse,
PermissionAssignment,
+ PermissionResource, PolicyRequest, PolicyType, RenameTableRequest,
ResourcePaths, ResourceType,
+ RevokePermissionRequest,
};
use paimon::catalog::{Function, Identifier};
use paimon::spec::{CommitKind, Partition, Snapshot};
@@ -86,6 +87,10 @@ struct MockState {
drop_policy_bodies: Vec<serde_json::Value>,
create_policy_error: Option<ErrorResponse>,
drop_policy_error: Option<ErrorResponse>,
+ auth_responses: HashMap<String, AuthTableQueryResponse>,
+ column_auth: HashMap<String, Vec<String>>,
+ uuid_after_auth: HashMap<String, String>,
+ uuid_after_calls: HashMap<String, (String, usize)>,
/// ECS metadata role name (for token loader testing)
ecs_role_name: Option<String>,
/// ECS metadata token (for token loader testing)
@@ -813,7 +818,7 @@ impl RESTServer {
Path((db, table)): Path<(String, String)>,
Extension(state): Extension<Arc<RESTServer>>,
) -> impl IntoResponse {
- let s = state.inner.lock().unwrap();
+ let mut s = state.inner.lock().unwrap();
let key = format!("{db}.{table}");
if s.no_permission_tables.contains(&key) {
@@ -826,6 +831,18 @@ impl RESTServer {
return (StatusCode::FORBIDDEN, Json(err)).into_response();
}
+ if let Some((uuid, remaining)) = s.uuid_after_calls.get_mut(&key) {
+ if *remaining == 0 {
+ let uuid = uuid.clone();
+ s.uuid_after_calls.remove(&key);
+ if let Some(t) = s.tables.get_mut(&key) {
+ t.id = Some(uuid);
+ }
+ } else {
+ *remaining -= 1;
+ }
+ }
+
if let Some(response) = s.tables.get(&key) {
return (StatusCode::OK, Json(response.clone())).into_response();
}
@@ -849,6 +866,78 @@ impl RESTServer {
(StatusCode::NOT_FOUND, Json(err)).into_response()
}
+ pub async fn auth_table_query(
+ Path((db, table)): Path<(String, String)>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ Json(request): Json<paimon::api::AuthTableQueryRequest>,
+ ) -> impl IntoResponse {
+ let s = state.inner.lock().unwrap();
+ let key = format!("{db}.{table}");
+
+ // Mirrors the reference server: a null select means the real schema
+ // fields, and any column outside the grant denies the query.
+ if let Some(allowed) = s.column_auth.get(&key) {
+ let requested = request.select.clone().unwrap_or_else(|| {
+ s.tables
+ .get(&key)
+ .and_then(|t| t.schema.as_ref())
+ .map(|schema| {
+ schema
+ .fields()
+ .iter()
+ .map(|f| f.name().to_string())
+ .collect()
+ })
+ .unwrap_or_default()
+ });
+ if let Some(denied) = requested.iter().find(|c|
!allowed.contains(c)) {
+ return (
+ StatusCode::FORBIDDEN,
+ Json(ErrorResponse::new(
+ Some("table".to_string()),
+ Some(denied.clone()),
+ Some(format!("no permission for column '{denied}'")),
+ Some(403),
+ )),
+ )
+ .into_response();
+ }
+ }
+
+ let response = s.auth_responses.get(&key).cloned().unwrap_or_default();
+ drop(s);
+ let mut s = state.inner.lock().unwrap();
+ if let Some(uuid) = s.uuid_after_auth.remove(&key) {
+ if let Some(existing) = s.tables.get_mut(&key) {
+ existing.id = Some(uuid);
+ }
+ }
+ (StatusCode::OK, Json(response)).into_response()
+ }
+
+ pub fn set_table_uuid_after_calls(
+ &self,
+ database: &str,
+ table: &str,
+ uuid: &str,
+ after: usize,
+ ) {
+ let mut s = self.inner.lock().unwrap();
+ s.uuid_after_calls
+ .insert(format!("{database}.{table}"), (uuid.to_string(), after));
+ }
+
+ pub fn set_table_uuid_after_auth(&self, database: &str, table: &str, uuid:
&str) {
+ let mut s = self.inner.lock().unwrap();
+ s.uuid_after_auth
+ .insert(format!("{database}.{table}"), uuid.to_string());
+ }
+
+ pub fn set_column_auth(&self, database: &str, table: &str, columns:
Vec<String>) {
+ let mut s = self.inner.lock().unwrap();
+ s.column_auth.insert(format!("{database}.{table}"), columns);
+ }
+
/// Handle DELETE /databases/:db/tables/:table - drop a table.
pub async fn drop_table(
Path((db, table)): Path<(String, String)>,
@@ -1843,10 +1932,13 @@ impl RESTServer {
});
let key = format!("{database}.{table}");
+ // A `t$branch_x` registration reports an id of its own: whether a real
+ // server shares the base table's is not something the client may
assume.
+ let uuid = table.to_string();
s.tables.insert(
key,
GetTableResponse::new(
- Some(table.to_string()),
+ Some(uuid),
Some(database.to_string()),
Some(table.to_string()),
Some(path.to_string()),
@@ -1858,6 +1950,42 @@ impl RESTServer {
);
}
+ pub fn clear_table_identity(&self, database: &str, table: &str) {
+ let mut s = self.inner.lock().unwrap();
+ if let Some(existing) =
s.tables.get_mut(&format!("{database}.{table}")) {
+ existing.id = None;
+ existing.schema_id = None;
+ }
+ }
+
+ pub fn set_table_uuid(&self, database: &str, table: &str, uuid: &str) {
+ let mut s = self.inner.lock().unwrap();
+ if let Some(existing) =
s.tables.get_mut(&format!("{database}.{table}")) {
+ existing.id = Some(uuid.to_string());
+ }
+ }
+
+ pub fn set_table_schema_id(
+ &self,
+ database: &str,
+ table: &str,
+ schema: paimon::spec::Schema,
+ schema_id: i64,
+ ) {
+ let mut s = self.inner.lock().unwrap();
+ let key = format!("{database}.{table}");
+ if let Some(existing) = s.tables.get_mut(&key) {
+ existing.schema_id = Some(schema_id);
+ existing.schema = Some(schema);
+ }
+ }
+
+ pub fn set_auth_response(&self, database: &str, table: &str, response:
AuthTableQueryResponse) {
+ let mut s = self.inner.lock().unwrap();
+ s.auth_responses
+ .insert(format!("{database}.{table}"), response);
+ }
+
/// Add a no-permission table to the server state.
pub fn add_no_permission_table(&self, database: &str, table: &str) {
let mut s = self.inner.lock().unwrap();
@@ -2229,6 +2357,10 @@ pub async fn start_mock_server(
&format!("{prefix}/databases/:db/functions/:function"),
get(RESTServer::get_function),
)
+ .route(
+ &format!("{prefix}/databases/:db/tables/:table/auth"),
+ post(RESTServer::auth_table_query),
+ )
.route(
&format!("{prefix}/tables/rename"),
post(RESTServer::rename_table),
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index c7ddad54..24f217c3 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -2672,6 +2672,409 @@ async fn
test_load_table_rejects_unknown_declared_type() {
);
}
+// Skipped on Windows for the same opendal `fs` StripPrefixError as the
+// blob-view regression above: it writes through FileSystemCatalog.
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_query_auth_unrestricted_user_can_read() {
+ let tmp = tempfile::tempdir().unwrap();
+ let warehouse = format!("file://{}", tmp.path().display());
+ let mut fs_options = Options::new();
+ fs_options.set(CatalogOptions::WAREHOUSE, &warehouse);
+ let fs_catalog = FileSystemCatalog::new(fs_options).expect("create
filesystem catalog");
+ fs_catalog
+ .create_database("default", true, HashMap::new())
+ .await
+ .unwrap();
+ let identifier = Identifier::new("default", "guarded");
+ let columns = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .build()
+ .unwrap();
+ fs_catalog
+ .create_table(&identifier, columns, false)
+ .await
+ .unwrap();
+ let plain = fs_catalog.get_table(&identifier).await.unwrap();
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![ArrowField::new(
+ "id",
+ ArrowDataType::Int32,
+ false,
+ )])),
+ vec![Arc::new(Int32Array::from(vec![1, 2, 3]))],
+ )
+ .unwrap();
+ write_batch(&plain, batch, "query-auth-fixture").await;
+
+ let ctx = setup_catalog(vec!["default"]).await;
+ let guarded_schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .option("query-auth.enabled", "true")
+ .build()
+ .unwrap();
+ ctx.server
+ .add_table_with_schema("default", "guarded", guarded_schema,
plain.location());
+
+ let table = ctx.catalog.get_table(&identifier).await.unwrap();
+ let read_builder = table.new_read_builder();
+ let plan = read_builder.new_scan().plan().await.unwrap();
+ assert!(
+ !plan.splits().is_empty(),
+ "the fixture must produce a split, or the read below proves nothing"
+ );
+
+ let batches = read_builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .expect("an unrestricted user must be allowed to read")
+ .try_collect::<Vec<_>>()
+ .await
+ .expect("and the rows must decode");
+ assert_eq!(
+ batches.iter().map(|b| b.num_rows()).sum::<usize>(),
+ 3,
+ "every written row must come back"
+ );
+}
+
+fn schema_of(columns: &[&str], options: &[(&str, &str)]) -> Schema {
+ let mut builder = Schema::builder();
+ for name in columns {
+ builder = builder.column(*name, DataType::Int(IntType::new()));
+ }
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().unwrap()
+}
+
+const GUARDED: &[(&str, &str)] = &[("query-auth.enabled", "true")];
+
+struct Guarded {
+ ctx: TestContext,
+ table: Table,
+ identifier: Identifier,
+ _tmp: tempfile::TempDir,
+}
+
+async fn guarded(name: &str, columns: &[&str]) -> Guarded {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let tmp = tempfile::tempdir().unwrap();
+ let path = format!("file://{}", tmp.path().display());
+ ctx.server
+ .add_table_with_schema("default", name, schema_of(columns, GUARDED),
&path);
+ let identifier = Identifier::new("default", name);
+ let table = ctx.catalog.get_table(&identifier).await.unwrap();
+ Guarded {
+ ctx,
+ table,
+ identifier,
+ _tmp: tmp,
+ }
+}
+
+async fn plan_err(table: &Table, why: &str) -> paimon::Error {
+ table
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .expect_err(why)
+}
+
+#[track_caller]
+fn assert_refused(err: paimon::Error) {
+ assert!(
+ matches!(err, paimon::Error::Unsupported { ref message }
+ if message.contains("query-auth.enabled")),
+ "{err:?}"
+ );
+}
+
+#[track_caller]
+fn assert_drifted(err: paimon::Error, what: &str) {
+ assert!(
+ matches!(err, paimon::Error::DataInvalid { ref message, .. }
+ if message.contains(what)),
+ "{err:?}"
+ );
+}
+
+fn restricted() -> paimon::api::AuthTableQueryResponse {
+ paimon::api::AuthTableQueryResponse {
+ filter: Some(vec!["{\"field\":\"id\"}".to_string()]),
+ column_masking: None,
+ }
+}
+
+#[tokio::test]
+async fn test_query_auth_restricted_user_is_refused_at_plan_time() {
+ let g = guarded("restricted", &["id"]).await;
+ g.ctx
+ .server
+ .set_auth_response("default", "restricted", restricted());
+
+ assert_refused(
+ plan_err(
+ &g.table,
+ "a restricted user must be refused before a plan exists",
+ )
+ .await,
+ );
+}
+
+#[tokio::test]
+async fn test_query_auth_refuses_a_stale_handle() {
+ let g = guarded("drifting", &["id"]).await;
+ g.ctx.server.set_table_schema_id(
+ "default",
+ "drifting",
+ schema_of(&["id", "extra"], GUARDED),
+ 7,
+ );
+
+ assert_drifted(
+ plan_err(&g.table, "a handle whose schema drifted must be
refused").await,
+ "now resolves to schema",
+ );
+}
+
+#[tokio::test]
+async fn test_query_auth_refuses_other_columns_under_the_same_schema_id() {
+ // Same id, other fields: the id alone does not say what the server rules
on.
+ let g = guarded("edited", &["id"]).await;
+ g.ctx
+ .server
+ .set_table_schema_id("default", "edited", schema_of(&["id", "extra"],
GUARDED), 0);
+
+ assert_drifted(
+ plan_err(
+ &g.table,
+ "a handle whose columns differ from the server's must be refused",
+ )
+ .await,
+ "serves other columns",
+ );
+}
+
+#[tokio::test]
+async fn test_query_auth_refuses_a_schema_replaced_copy() {
+ // A caller can restore a dropped column under the current schema id; the
+ // copy is no longer the handle the catalog loaded.
+ let g = guarded("replaced", &["id"]).await;
+ let forged = g
+ .table
+ .copy_with_resolved_schema(
+ paimon::spec::TableSchema::new(0, &schema_of(&["id", "secret"],
GUARDED)),
+ "main",
+ )
+ .unwrap();
+
+ assert_refused(plan_err(&forged, "a schema-replaced copy must not
plan").await);
+}
+
+#[tokio::test]
+async fn test_query_auth_refuses_a_recreated_table() {
+ let g = guarded("recreated", &["id"]).await;
+ g.ctx
+ .server
+ .set_table_uuid("default", "recreated", "uuid-of-the-replacement");
+
+ assert_drifted(
+ plan_err(
+ &g.table,
+ "a re-created table must not reuse this handle's grant",
+ )
+ .await,
+ "now resolves to uuid",
+ );
+}
+
+#[tokio::test]
+async fn test_query_auth_refuses_a_table_recreated_while_planning() {
+ let g = guarded("planned", &["id"]).await;
+ g.ctx
+ .server
+ .set_table_uuid_after_calls("default", "planned",
"uuid-of-the-replacement", 2);
+
+ assert_drifted(
+ plan_err(&g.table, "the files just planned belong to the
replacement").await,
+ "now resolves to uuid",
+ );
+}
+
+#[tokio::test]
+async fn test_query_auth_refuses_a_table_recreated_mid_exchange() {
+ let g = guarded("swapped", &["id"]).await;
+ g.ctx
+ .server
+ .set_table_uuid_after_auth("default", "swapped",
"uuid-after-the-exchange");
+
+ assert_drifted(
+ plan_err(&g.table, "a table replaced mid-exchange must be
refused").await,
+ "now resolves to uuid",
+ );
+}
+
+#[tokio::test]
+async fn test_query_auth_refuses_a_server_that_reports_no_identity() {
+ let g = guarded("anonymous", &["id"]).await;
+ g.ctx.server.clear_table_identity("default", "anonymous");
+
+ assert_drifted(
+ plan_err(
+ &g.table,
+ "a check that cannot establish the table has not checked anything",
+ )
+ .await,
+ "nothing the server reports",
+ );
+}
+
+#[tokio::test]
+async fn test_query_auth_user_granted_all_business_columns_can_read() {
+ let g = guarded("granted", &["id", "name"]).await;
+ g.ctx.server.set_column_auth(
+ "default",
+ "granted",
+ vec!["id".to_string(), "name".to_string()],
+ );
+
+ g.table
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .expect("a user granted every column must be authorized");
+}
+
+#[tokio::test]
+async fn
test_query_auth_is_not_weakened_by_a_table_recreated_under_the_same_name() {
+ let g = guarded("guarded", &["id"]).await;
+ g.ctx
+ .server
+ .set_table_schema_id("default", "guarded", schema_of(&["id"], &[]), 0);
+
+ let err = g
+ .table
+ .new_read_builder()
+ .new_scan()
+ .with_scan_all_files()
+ .plan()
+ .await
+ .expect_err("the answer is now about a different table over the same
files");
+ assert_refused(err);
+}
+
+#[tokio::test]
+async fn
test_query_auth_allows_a_nested_projection_but_not_an_extra_nested_field() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let tmp = tempfile::tempdir().unwrap();
+ let path = format!("file://{}", tmp.path().display());
+ let nested = |names: &[&str]| {
+ let children = names
+ .iter()
+ .enumerate()
+ .map(|(i, name)| {
+ paimon::spec::DataField::new(
+ i as i32 + 1,
+ name.to_string(),
+ DataType::Int(IntType::new()),
+ )
+ })
+ .collect();
+ DataType::Row(paimon::spec::RowType::new(children))
+ };
+ let served = Schema::builder()
+ .column("info", nested(&["a", "b"]))
+ .option("query-auth.enabled", "true")
+ .build()
+ .unwrap();
+ ctx.server
+ .add_table_with_schema("default", "nested", served, &path);
+
+ let table = ctx
+ .catalog
+ .get_table(&Identifier::new("default", "nested"))
+ .await
+ .unwrap();
+ let plan = table.new_read_builder().new_scan().plan().await.unwrap();
+ let read_with = |info: DataType| {
+ let field =
+ paimon::spec::DataField::new(table.schema().fields()[0].id(),
"info".to_string(), info);
+ let mut builder = table.new_read_builder();
+ builder.with_read_type(vec![field]);
+ builder.new_read().unwrap().to_arrow(plan.splits())
+ };
+
+ // Reading a subset of the authorized children is a projection.
+ assert!(read_with(nested(&["a"])).is_ok());
+ // Reading one the server never ruled on is not.
+ let Err(err) = read_with(nested(&["a", "b", "hidden"])) else {
+ panic!("a nested child the server never ruled on must be refused")
+ };
+ assert_refused(err);
+}
+
+#[tokio::test]
+async fn test_a_decorated_name_loads_and_only_query_auth_refuses_it() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let tmp = tempfile::tempdir().unwrap();
+ let path = format!("file://{}", tmp.path().display());
+ // A literal name the validator permits and the server stores as is: the
+ // catalog hands it back, as on main.
+ ctx.server
+ .add_table_with_schema("default", "plain$files", schema_of(&["id"],
&[]), &path);
+ for name in ["guarded$branch_dev", "guarded$files"] {
+ ctx.server
+ .add_table_with_schema("default", name, schema_of(&["id"],
GUARDED), &path);
+ }
+ ctx.catalog
+ .get_table(&Identifier::new("default", "plain$files"))
+ .await
+ .expect("an ordinary table's name is the server's business");
+ // The view such a name addresses is not what the server rules on.
+ for name in ["guarded$branch_dev", "guarded$files"] {
+ let table = ctx
+ .catalog
+ .get_table(&Identifier::new("default", name))
+ .await
+ .unwrap();
+ assert_refused(plan_err(&table, name).await);
+ }
+}
+
+#[tokio::test]
+async fn test_query_auth_refuses_an_assembled_handle() {
+ let g = guarded("guarded", &["id"]).await;
+ let elsewhere = tempfile::tempdir().unwrap();
+ for (schema, location) in [
+ (g.table.schema().clone(), g.table.location().to_string()),
+ (
+ paimon::spec::TableSchema::new(
+ g.table.schema().id(),
+ &schema_of(&["id", "dropped"], GUARDED),
+ ),
+ g.table.location().to_string(),
+ ),
+ (
+ g.table.schema().clone(),
+ format!("file://{}", elsewhere.path().display()),
+ ),
+ ] {
+ let assembled = paimon::table::Table::new(
+ g.table.file_io().clone(),
+ g.identifier.clone(),
+ location,
+ schema,
+ g.table.rest_env().cloned(),
+ );
+ assert_refused(plan_err(&assembled, "an assembled handle carries no
session").await);
+ }
+}
+
#[tokio::test]
async fn test_rest_catalog_manages_permissions_end_to_end() {
let ctx = setup_catalog(vec!["default"]).await;
