diff --git a/crates/integrations/datafusion/src/partition_count_pushdown.rs b/crates/integrations/datafusion/src/partition_count_pushdown.rs index 58aad5344..0dc73a53b 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>, output_schema: SchemaRef, source: Arc, + unpinned_source: Arc, table_name: TableReference, filters: Vec, state: Arc, @@ -362,6 +371,7 @@ impl PartitionRowCountStream { provider: &PartitionRowCountProvider, table: Option, source: Arc, + unpinned_source: Arc, projection: Option>, state: SessionState, ) -> DFResult { @@ -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> { - let source_schema = self.source.schema(); + fn fallback_stream( + &self, + plan: Arc, + context: Arc, + ) -> DFResult { + 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, + ) -> DFResult> { + let source_schema = source.schema(); let partition_indices = self .partition_fields .iter() @@ -485,7 +519,7 @@ impl PartitionRowCountStream { .collect::, _>>()?; 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 943f861cf..0a595f021 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 114b89ab9..7ee19b151 100644 --- a/crates/paimon/src/spec/core_options.rs +++ b/crates/paimon/src/spec/core_options.rs @@ -1063,7 +1063,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 6f0e80834..9cee8b896 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -638,22 +638,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 d81f1f3dd..f63b9c00a 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 { - 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 { + // 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 125a2a072..3eb95274e 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 { 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 0b343e366..be8d7fd34 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -107,6 +107,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; @@ -221,6 +222,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, rest_env: Option, /// True when this table copy was switched to a historical schema by /// [`Table::copy_with_time_travel`]. Such a copy is read-only. @@ -250,6 +254,7 @@ impl Table { schema_manager, branch, branch_reference: false, + query_auth_session: None, rest_env, time_traveled: false, travel_snapshot: None, @@ -290,6 +295,7 @@ impl Table { schema_manager, branch, branch_reference, + query_auth_session: None, rest_env: None, time_traveled: false, travel_snapshot: None, @@ -374,6 +380,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 { + // 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>> { + 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 { + 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() @@ -483,6 +558,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 { @@ -517,6 +593,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() }) } @@ -724,6 +802,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, @@ -758,6 +837,32 @@ 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 table = query_auth_table(); + Table { + rest_env: Some(RESTEnv::new( + table.identifier.clone(), + "uuid-1".to_string(), + api, + options, + false, + None, + )), + ..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 8b998bd80..36f460ea4 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 = None; let mut seen_buckets: HashSet = 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 000000000..fb24d3bae --- /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, +) -> 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>, + written: Option>, + ) -> 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| 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 751833658..fa95ddb82 100644 --- a/crates/paimon/src/table/read_builder.rs +++ b/crates/paimon/src/table/read_builder.rs @@ -549,10 +549,20 @@ impl<'a> PaimonReadBuilder<'a> { /// Create a table read for consuming splits (e.g. from a scan plan). pub fn new_read(&self) -> Result> { - // 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. - let core_options = self.table.schema.core_options(); - core_options.ensure_read_authorized()?; + // The declared type needs no grant; only query-auth moved to `to_arrow`. + self.table + .schema + .core_options() + .ensure_type_paimon_served(&self.table.identifier().full_name())?; + // A handle no catalog loaded holds no grant; refused here too, as bindings + // skip `to_arrow` for an empty split list. + if self.table.schema.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, @@ -759,6 +769,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 { @@ -1002,14 +1025,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`. @@ -1017,7 +1059,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 af810f088..687935339 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; @@ -116,6 +117,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>, + ) -> Result { + 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 { + let response = self.api.get_table(&self.identifier).await?; + let name = self.identifier.full_name(); + let same = |what: &str, loaded: String, now: Option| 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::>() { + 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 @@ -168,8 +229,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, @@ -265,9 +324,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 f96a4f47a..ab501eebe 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>, + /// 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>) -> 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> { + 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> { + 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> { + 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>) -> 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 3e4b9c083..3728edff6 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)?; @@ -6243,6 +6245,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 c7516df40..ee0d3a0fc 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 { 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 { - 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 { - 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 { - 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) { + 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 { + // 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 { 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::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 3358203ca..8b860916d 100644 --- a/crates/paimon/src/table/table_scan.rs +++ b/crates/paimon/src/table/table_scan.rs @@ -1275,15 +1275,16 @@ impl<'a> TableScan<'a> { } pub async fn plan(&self) -> crate::Result { + // 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, } } @@ -1506,45 +1507,110 @@ 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 { - self.ensure_query_auth_allowed()?; + 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.ensure_query_auth_allowed()?; + 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>> { + 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())?; + + // 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 ensure_query_auth_allowed(&self) -> crate::Result<()> { CoreOptions::new(self.table.schema().options()).ensure_read_authorized() } @@ -2779,6 +2845,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 893c61ca6..68dadf346 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, create_policy_error: Option, drop_policy_error: Option, + auth_responses: HashMap, + column_auth: HashMap>, + uuid_after_auth: HashMap, + uuid_after_calls: HashMap, /// ECS metadata role name (for token loader testing) ecs_role_name: Option, /// ECS metadata token (for token loader testing) @@ -813,7 +818,7 @@ impl RESTServer { Path((db, table)): Path<(String, String)>, Extension(state): Extension>, ) -> 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>, + Json(request): Json, + ) -> 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) { + 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 e2dd53ecb..e194c8919 100644 --- a/crates/paimon/tests/rest_catalog_test.rs +++ b/crates/paimon/tests/rest_catalog_test.rs @@ -2574,6 +2574,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::>() + .await + .expect("and the rows must decode"); + assert_eq!( + batches.iter().map(|b| b.num_rows()).sum::(), + 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;