[table] Fail closed when reading a query-auth.enabled table

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);
             }

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;

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"`).

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

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<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()
@@ -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"
+        );
+    }
 }

crates/paimon/src/table/mod.rs

@@ -278,3 +278,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,
+    )
+}

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();

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"
+        );
+    }
 }

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"
+        );
+    }
 }

crates/paimon/src/table/vector_search_builder.rs

@@ -95,6 +95,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()
@@ -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,