This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new aba5b5b [table] Fail closed when reading a query-auth.enabled table
(#447)
aba5b5b is described below
commit aba5b5bcdc7d29265c57cd52adb5b6a9d488fcae
Author: Jiajia Li <[email protected]>
AuthorDate: Sat Jul 4 23:25:27 2026 +0800
[table] Fail closed when reading a query-auth.enabled table (#447)
---
.../datafusion/src/system_tables/mod.rs | 4 ++
.../integrations/datafusion/tests/system_tables.rs | 35 ++++++++++++++++++
crates/paimon/src/spec/core_options.rs | 27 ++++++++++++++
crates/paimon/src/spec/schema.rs | 10 ++++-
.../paimon/src/table/full_text_search_builder.rs | 16 ++++++++
crates/paimon/src/table/mod.rs | 25 +++++++++++++
crates/paimon/src/table/read_builder.rs | 32 +++++++++++++++-
crates/paimon/src/table/table_read.rs | 18 +++++++++
crates/paimon/src/table/table_scan.rs | 43 ++++++++++++++++++++++
crates/paimon/src/table/vector_search_builder.rs | 16 ++++++++
10 files changed, 222 insertions(+), 4 deletions(-)
diff --git a/crates/integrations/datafusion/src/system_tables/mod.rs
b/crates/integrations/datafusion/src/system_tables/mod.rs
index 0305d2f..c46d3cd 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 ad4805d..be1d13e 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<RecordBatch> {
.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 f11298f..f5f71c5 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 777b62f..75b35c9 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<String, String>) -> 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<String, String>) ->
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 b24e82e..616d1ed 100644
--- a/crates/paimon/src/table/full_text_search_builder.rs
+++ b/crates/paimon/src/table/full_text_search_builder.rs
@@ -96,6 +96,8 @@ impl<'a> FullTextSearchBuilder<'a> {
///
/// Reference: `FullTextSearchBuilder.executeLocal()`
pub async fn execute(&self) -> crate::Result<Vec<RowRange>> {
+ // 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()
@@ -542,4 +544,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 19e9f00..4a59948 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -282,3 +282,28 @@ pub type ArrowRecordBatchStream = BoxStream<'static,
Result<RecordBatch>>;
pub(crate) fn find_field_id_by_name(fields: &[DataField], name: &str) ->
Option<i32> {
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 dbf22ca..1b21d24 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<TableRead<'a>> {
+ // Fail closed at read construction so bindings that short-circuit
before
+ // `to_arrow` (e.g. an empty-splits fast path) can't bypass the guard.
+
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 6f71455..853ce9c 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<ArrowRecordBatchStream> {
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<i64>) ->
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 57f21d8..93a1c35 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<Plan> {
+ 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<Option<HashSet<i32>>> {
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 2ca9993..fc0d2b7 100644
--- a/crates/paimon/src/table/vector_search_builder.rs
+++ b/crates/paimon/src/table/vector_search_builder.rs
@@ -105,6 +105,8 @@ impl<'a> VectorSearchBuilder<'a> {
}
pub async fn execute(&self) -> crate::Result<Vec<RowRange>> {
+ // 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()
@@ -1282,6 +1284,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,
