diff --git a/crates/integrations/datafusion/src/system_tables/mod.rs b/crates/integrations/datafusion/src/system_tables/mod.rs index 0305d2f8..c46d3cd7 100644 --- a/crates/integrations/datafusion/src/system_tables/mod.rs +++ b/crates/integrations/datafusion/src/system_tables/mod.rs @@ -139,6 +139,10 @@ pub(crate) async fn load( let identifier = Identifier::new(database, base.clone()); match catalog.get_table(&identifier).await { Ok(table) => { + // Fail closed: system tables expose file metadata the client can't authorize. + paimon::spec::CoreOptions::new(table.schema().options()) + .ensure_read_authorized() + .map_err(to_datafusion_error)?; if system_name.eq_ignore_ascii_case("partitions") { return partitions::build(catalog, identifier, table).map(Some); } diff --git a/crates/integrations/datafusion/tests/system_tables.rs b/crates/integrations/datafusion/tests/system_tables.rs index ad4805d0..be1d13ec 100644 --- a/crates/integrations/datafusion/tests/system_tables.rs +++ b/crates/integrations/datafusion/tests/system_tables.rs @@ -72,6 +72,41 @@ async fn run_sql(ctx: &SQLContext, sql: &str) -> Vec { .unwrap_or_else(|e| panic!("Failed to execute `{sql}`: {e}")) } +// Error text of `sql`, whether it surfaces at planning or execution. +async fn query_error(ctx: &SQLContext, sql: &str) -> String { + match ctx.sql(sql).await { + Err(e) => e.to_string(), + Ok(df) => df + .collect() + .await + .expect_err(&format!("`{sql}` must fail")) + .to_string(), + } +} + +#[tokio::test] +async fn test_query_auth_table_fails_closed() { + let (ctx, _catalog, _tmp) = create_context().await; + run_sql( + &ctx, + "CREATE TABLE paimon.default.qa (id INT) WITH ('query-auth.enabled' = 'true')", + ) + .await; + + // Data reads and data-derived system tables must all fail closed. + for sql in [ + "SELECT * FROM paimon.default.qa", + "SELECT * FROM paimon.default.qa$manifests", + "SELECT * FROM paimon.default.qa$table_indexes", + ] { + let err = query_error(&ctx, sql).await; + assert!( + err.contains("query-auth.enabled"), + "`{sql}` should fail closed, got: {err}" + ); + } +} + #[tokio::test] async fn test_options_system_table() { let (ctx, catalog, _tmp) = create_context().await; diff --git a/crates/paimon/src/spec/core_options.rs b/crates/paimon/src/spec/core_options.rs index f11298f7..f5f71c51 100644 --- a/crates/paimon/src/spec/core_options.rs +++ b/crates/paimon/src/spec/core_options.rs @@ -18,6 +18,7 @@ use std::collections::{HashMap, HashSet}; const DELETION_VECTORS_ENABLED_OPTION: &str = "deletion-vectors.enabled"; +pub(crate) const QUERY_AUTH_ENABLED_OPTION: &str = "query-auth.enabled"; const DATA_EVOLUTION_ENABLED_OPTION: &str = "data-evolution.enabled"; const GLOBAL_INDEX_ENABLED_OPTION: &str = "global-index.enabled"; const GLOBAL_INDEX_SEARCH_MODE_OPTION: &str = "global-index.search-mode"; @@ -207,6 +208,32 @@ impl<'a> CoreOptions<'a> { .unwrap_or(false) } + /// Whether `query-auth.enabled` is set. + /// + /// When set, the server enforces a per-user row filter / column masking that this client + /// can't yet apply, so read paths fail closed (see `ensure_read_authorized`). + pub fn query_auth_enabled(&self) -> bool { + self.options + .get(QUERY_AUTH_ENABLED_OPTION) + .map(|value| value.eq_ignore_ascii_case("true")) + .unwrap_or(false) + } + + /// Fail closed when `query-auth.enabled` is set: this client can't enforce the row + /// filter / column masking, so refuse to read. Call at every read boundary (build, + /// plan, materialize) so no binding fast-path can bypass it. + pub fn ensure_read_authorized(&self) -> crate::Result<()> { + if self.query_auth_enabled() { + return Err(crate::Error::Unsupported { + message: "reading a table with 'query-auth.enabled' = true is not supported: \ + the Rust client cannot yet enforce its row-level auth filter / column \ + masking, so it refuses to read to avoid returning unfiltered data" + .to_string(), + }); + } + Ok(()) + } + /// Returns the user-specified sequence field names, if configured. /// When set, the values of these columns are used as `_SEQUENCE_NUMBER` instead of auto-increment. /// Multiple fields can be comma-separated (e.g. `"col_a,col_b"`). diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index 777b62f3..75b35c93 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -16,7 +16,8 @@ // under the License. use crate::spec::core_options::{ - first_row_supports_changelog_producer, CoreOptions, BUCKET_KEY_OPTION, SEQUENCE_FIELD_OPTION, + first_row_supports_changelog_producer, CoreOptions, BUCKET_KEY_OPTION, + QUERY_AUTH_ENABLED_OPTION, SEQUENCE_FIELD_OPTION, }; use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType}; use crate::spec::{ @@ -130,7 +131,12 @@ impl TableSchema { } /// Create a copy of this schema with extra options merged in. - pub fn copy_with_options(&self, extra: HashMap) -> Self { + /// + /// A stored `query-auth.enabled = true` can't be turned off by a dynamic override. + pub fn copy_with_options(&self, mut extra: HashMap) -> Self { + if CoreOptions::new(&self.options).query_auth_enabled() { + extra.insert(QUERY_AUTH_ENABLED_OPTION.to_string(), "true".to_string()); + } let mut new_schema = self.clone(); new_schema.options.extend(extra); new_schema diff --git a/crates/paimon/src/table/full_text_search_builder.rs b/crates/paimon/src/table/full_text_search_builder.rs index 04ffe214..8ac14641 100644 --- a/crates/paimon/src/table/full_text_search_builder.rs +++ b/crates/paimon/src/table/full_text_search_builder.rs @@ -93,6 +93,8 @@ impl<'a> FullTextSearchBuilder<'a> { /// /// Reference: `FullTextSearchBuilder.executeLocal()` pub async fn execute(&self) -> crate::Result> { + // Fail closed: returns data-derived row ranges outside `TableScan`/`TableRead`. + CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?; let text_column = self.text_column .as_deref() @@ -516,4 +518,18 @@ mod tests { )])); RecordBatch::try_new(schema, vec![Arc::new(StringArray::from(values))]).unwrap() } + + #[tokio::test] + async fn test_execute_fails_closed_when_query_auth_enabled() { + let table = crate::table::query_auth_table(); + let err = table + .new_full_text_search_builder() + .execute() + .await + .unwrap_err(); + assert!( + matches!(err, crate::Error::Unsupported { ref message } if message.contains("query-auth.enabled")), + "full-text search must fail closed for a query-auth table" + ); + } } diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index b961f76d..f9d2afb7 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -278,3 +278,28 @@ pub type ArrowRecordBatchStream = BoxStream<'static, Result>; 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 minimal table with `query-auth.enabled = true`, for the fail-closed read guard. +#[cfg(test)] +pub(crate) fn query_auth_table() -> Table { + use crate::catalog::Identifier; + use crate::io::FileIOBuilder; + use crate::spec::{DataType, IntType, Schema, TableSchema}; + + let file_io = FileIOBuilder::new("file").build().unwrap(); + let table_schema = TableSchema::new( + 0, + &Schema::builder() + .column("id", DataType::Int(IntType::new())) + .option("query-auth.enabled", "true") + .build() + .unwrap(), + ); + Table::new( + file_io, + Identifier::new("default", "auth_t"), + "/tmp/test-query-auth-table".to_string(), + table_schema, + None, + ) +} diff --git a/crates/paimon/src/table/read_builder.rs b/crates/paimon/src/table/read_builder.rs index dbf22cac..1b21d242 100644 --- a/crates/paimon/src/table/read_builder.rs +++ b/crates/paimon/src/table/read_builder.rs @@ -228,6 +228,9 @@ impl<'a> ReadBuilder<'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. + CoreOptions::new(self.table.schema.options()).ensure_read_authorized()?; let read_type = match &self.projected_fields { None => self.table.schema.fields().to_vec(), Some(projected) => self.resolve_projected_fields(projected)?, @@ -334,10 +337,10 @@ mod tests { use crate::spec::{ BinaryRow, DataType, IntType, Predicate, PredicateBuilder, Schema, TableSchema, VarCharType, }; - use crate::table::{DataSplitBuilder, Table}; + use crate::table::{query_auth_table, DataSplitBuilder, Table}; use arrow_array::{Int32Array, RecordBatch}; use futures::TryStreamExt; - use std::collections::HashSet; + use std::collections::{HashMap, HashSet}; use std::fs; use tempfile::tempdir; use test_utils::{local_file_path, test_data_file, write_int_parquet_file}; @@ -398,6 +401,31 @@ 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(); + 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" + ); + } + + #[test] + fn test_dynamic_option_cannot_disable_query_auth() { + // Copying the table with the option off must not weaken a stored `true`. + let table = query_auth_table().copy_with_options(HashMap::from([( + "query-auth.enabled".to_string(), + "false".to_string(), + )])); + let err = table.new_read_builder().new_read().unwrap_err(); + assert!( + matches!(err, crate::Error::Unsupported { ref message } if message.contains("query-auth.enabled")), + "a dynamic override must not disable query-auth" + ); + } + #[test] fn test_projected_read_field_ids_uses_projection_ids() { let table = simple_table(); diff --git a/crates/paimon/src/table/table_read.rs b/crates/paimon/src/table/table_read.rs index 6f71455c..853ce9c4 100644 --- a/crates/paimon/src/table/table_read.rs +++ b/crates/paimon/src/table/table_read.rs @@ -74,6 +74,8 @@ impl<'a> TableRead<'a> { pub fn to_arrow(&self, data_splits: &[DataSplit]) -> crate::Result { let has_primary_keys = !self.table.schema.primary_keys().is_empty(); let core_options = CoreOptions::new(self.table.schema.options()); + // Fail closed for a direct `TableRead` (bypassing `ReadBuilder::new_read`). + core_options.ensure_read_authorized()?; let merge_engine = core_options.merge_engine()?; // PK table with Deduplicate engine: splits that may hold multiple @@ -237,6 +239,7 @@ mod tests { use super::*; use crate::spec::stats::BinaryTableStats; use crate::spec::{BinaryRow, DataFileMeta}; + use crate::table::query_auth_table; use crate::table::source::DataSplitBuilder; fn file(name: &str, level: i32, delete_row_count: Option) -> DataFileMeta { @@ -298,4 +301,19 @@ mod tests { let dv_compacted = split(vec![file("a", 5, None)], false); assert!(!pk_split_needs_merge(&dv_compacted, true)); } + + #[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()); + assert!( + matches!( + read.to_arrow(&[]), + 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 57f21d8d..93a1c35a 100644 --- a/crates/paimon/src/table/table_scan.rs +++ b/crates/paimon/src/table/table_scan.rs @@ -634,6 +634,7 @@ impl<'a> TableScan<'a> { /// /// Reference: [TimeTravelUtil.tryTravelToSnapshot](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java) pub async fn plan(&self) -> crate::Result { + self.ensure_query_auth_allowed()?; let data_evolution_read_field_ids = self.projected_read_field_ids()?; let snapshot = match self.resolve_snapshot().await? { Some(snapshot) => snapshot, @@ -645,6 +646,7 @@ impl<'a> TableScan<'a> { /// 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 data_evolution_read_field_ids = self.projected_read_field_ids()?; let mut trace = ScanTrace { limit: self.limit, @@ -665,6 +667,13 @@ impl<'a> TableScan<'a> { Ok((plan, trace)) } + /// 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. + fn ensure_query_auth_allowed(&self) -> crate::Result<()> { + CoreOptions::new(self.table.schema().options()).ensure_read_authorized() + } + fn projected_read_field_ids(&self) -> crate::Result>> { super::read_builder::projected_read_field_ids( self.table.identifier().full_name(), @@ -2565,4 +2574,38 @@ mod tests { let bucket = *buckets.iter().next().unwrap(); assert!((0..8).contains(&bucket)); } + + #[tokio::test] + async fn test_plan_fails_closed_when_query_auth_enabled() { + // Every scan-planning path must fail closed, including `with_scan_all_files` + // (read-facing system tables like `files` use it to expose metadata). + let table = crate::table::query_auth_table(); + let rb = table.new_read_builder(); + for scan in [rb.new_scan(), rb.new_scan().with_scan_all_files()] { + let err = scan.plan().await.unwrap_err(); + assert!( + matches!(err, crate::Error::Unsupported { ref message } if message.contains("query-auth.enabled")), + "scan planning must fail closed (scan_all_files or not)" + ); + } + } + + #[tokio::test] + async fn test_dynamic_option_cannot_disable_query_auth_at_plan() { + // Copying the table with the option off must not weaken a stored `true`. + let table = + crate::table::query_auth_table().copy_with_options(std::collections::HashMap::from([ + ("query-auth.enabled".to_string(), "false".to_string()), + ])); + let err = table + .new_read_builder() + .new_scan() + .plan() + .await + .unwrap_err(); + 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/vector_search_builder.rs b/crates/paimon/src/table/vector_search_builder.rs index 1ebd422c..97aea9e1 100644 --- a/crates/paimon/src/table/vector_search_builder.rs +++ b/crates/paimon/src/table/vector_search_builder.rs @@ -95,6 +95,8 @@ impl<'a> VectorSearchBuilder<'a> { } pub async fn execute(&self) -> crate::Result> { + // Fail closed: returns data-derived row ranges outside `TableScan`/`TableRead`. + CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?; let vector_column = self.vector_column .as_deref() @@ -922,6 +924,20 @@ mod tests { ); } + #[tokio::test] + async fn test_execute_fails_closed_when_query_auth_enabled() { + let table = crate::table::query_auth_table(); + let err = table + .new_vector_search_builder() + .execute() + .await + .unwrap_err(); + assert!( + matches!(err, crate::Error::Unsupported { ref message } if message.contains("query-auth.enabled")), + "vector search must fail closed for a query-auth table" + ); + } + fn make_lumina_entry( file_name: &str, index_type: &str,