fix[auth] Fail closed on query-auth tables in write, index and metadata paths

crates/paimon/src/catalog/partition_listing.rs

@@ -32,6 +32,8 @@ use crate::Result;
 /// Scan a table's manifest entries and aggregate them into [`Partition`] rows,
 /// matching the shape catalogs would otherwise return from a metastore.
 pub async fn list_partitions_from_file_system(table: &Table) -> Result<Vec<Partition>> {
+    // Manifests carry partition values and per-column stats.
+    crate::spec::CoreOptions::new(table.schema().options()).ensure_read_authorized()?;
     let file_io = table.file_io();
     let snapshot_sm = table.snapshot_manager();
     let manifest_sm = SnapshotManager::new(file_io.clone(), table.location().to_string());
@@ -113,3 +115,19 @@ pub async fn list_partitions_from_file_system(table: &Table) -> Result<Vec<Parti
     }
     Ok(result)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_query_auth_table_refuses_partition_listing() {
+        let table = crate::table::query_auth_table();
+        let err = list_partitions_from_file_system(&table).await.unwrap_err();
+        assert!(
+            matches!(err, crate::Error::Unsupported { ref message }
+                if message.contains("query-auth.enabled")),
+            "listing partitions on a query-auth.enabled table must fail closed, got {err:?}"
+        );
+    }
+}

crates/paimon/src/table/btree_global_index_build_builder.rs

@@ -83,6 +83,9 @@ impl<'a> BTreeGlobalIndexBuildBuilder<'a> {
     }
 
     pub async fn execute(&self) -> Result<usize> {
+        // Building the index scans the table's rows.
+        CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+
         self.table.ensure_not_branch_reference_for_write()?;
 
         let index_type = normalize_sorted_global_index_type(&self.index_type).ok_or_else(|| {

crates/paimon/src/table/bucket_assigner_cross.rs

@@ -71,6 +71,9 @@ impl GlobalPartitionIndex {
         target_bucket_row_number: i64,
         merge_engine: MergeEngine,
     ) -> Result<Self> {
+        // The cross-partition index reads every primary key.
+        crate::spec::CoreOptions::new(table.schema().options()).ensure_read_authorized()?;
+
         let mut key_to_location: HashMap<Vec<u8>, (Vec<u8>, i32)> = HashMap::new();
         let mut bucket_row_counts: HashMap<(Vec<u8>, i32), i64> = HashMap::new();
 

crates/paimon/src/table/cow_writer.rs

@@ -206,6 +206,9 @@ impl CopyOnWriteMergeWriter {
     /// Rewrite affected files and produce CommitMessages.
     #[must_use = "commit messages must be passed to TableCommit"]
     pub async fn prepare_commit(self) -> Result<Vec<CommitMessage>> {
+        // A copy-on-write rewrite reads the rows it replaces.
+        CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+
         if self.affected_files.is_empty() {
             return Ok(Vec::new());
         }

crates/paimon/src/table/data_evolution_writer.rs

@@ -159,6 +159,9 @@ impl DataEvolutionWriter {
     /// Returns `CommitMessage`s for the caller to commit via [`TableCommit`](super::TableCommit).
     #[must_use = "commit messages must be passed to TableCommit"]
     pub async fn prepare_commit(self) -> Result<Vec<CommitMessage>> {
+        // A row-id update reads the original rows it rewrites.
+        CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+
         let total_matched: usize = self.matched_batches.iter().map(|b| b.num_rows()).sum();
         if total_matched == 0 {
             return Ok(Vec::new());
@@ -455,6 +458,9 @@ impl DataEvolutionDeleteWriter {
 
     #[must_use = "commit messages must be passed to TableCommit"]
     pub async fn prepare_commit(mut self) -> Result<Vec<CommitMessage>> {
+        // A row-id delete reads the files it rewrites.
+        CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+
         dedup_i64_in_place(&mut self.row_ids);
         if self.row_ids.is_empty() {
             return Ok(Vec::new());

crates/paimon/src/table/global_index_drop_builder.rs

@@ -50,6 +50,9 @@ impl<'a> GlobalIndexDropBuilder<'a> {
     }
 
     pub async fn execute(&self) -> Result<usize> {
+        // Dropping an index reads the index manifest.
+        crate::spec::CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+
         self.table.ensure_not_branch_reference_for_write()?;
 
         let index_type =
@@ -170,6 +173,53 @@ mod tests {
     use chrono::{DateTime, Utc};
     use indexmap::IndexMap;
 
+    #[tokio::test]
+    async fn test_query_auth_table_refuses_index_builds_and_drops() {
+        let table = crate::table::query_auth_table();
+
+        let refused = |err: crate::Error, what: &str| {
+            assert!(
+                matches!(err, crate::Error::Unsupported { ref message }
+                    if message.contains("query-auth.enabled")),
+                "{what} on a query-auth.enabled table must fail closed, got {err:?}"
+            );
+        };
+
+        refused(
+            table
+                .new_global_index_drop_builder()
+                .with_index_type("btree")
+                .execute()
+                .await
+                .unwrap_err(),
+            "dropping a global index",
+        );
+        refused(
+            table
+                .new_btree_global_index_build_builder()
+                .execute()
+                .await
+                .unwrap_err(),
+            "building a BTree global index",
+        );
+        refused(
+            table
+                .new_vindex_index_build_builder("vector")
+                .execute()
+                .await
+                .unwrap_err(),
+            "building a vector index",
+        );
+        refused(
+            table
+                .new_lumina_index_build_builder()
+                .execute()
+                .await
+                .unwrap_err(),
+            "building a lumina index",
+        );
+    }
+
     fn test_table(table_path: &str) -> Table {
         let schema = Schema::builder()
             .column("id", DataType::Int(IntType::new()))

crates/paimon/src/table/lumina_index_build_builder.rs

@@ -71,6 +71,9 @@ impl<'a> LuminaIndexBuildBuilder<'a> {
     }
 
     pub async fn execute(&self) -> Result<usize> {
+        // Building the index scans the table's rows.
+        CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+
         self.table.ensure_not_branch_reference_for_write()?;
 
         if !is_lumina_index_type(&self.index_type) {

crates/paimon/src/table/partition_stat.rs

@@ -64,6 +64,8 @@ impl Table {
     ///
     /// Returns an empty Vec when the table has no snapshots yet.
     pub async fn partition_stats(&self) -> crate::Result<Vec<PartitionStat>> {
+        // Manifests carry partition values and per-column stats.
+        CoreOptions::new(self.schema().options()).ensure_read_authorized()?;
         let sm = SnapshotManager::new(self.file_io().clone(), self.location().to_string());
         let snapshot = match sm.get_latest_snapshot().await? {
             Some(s) => s,
@@ -189,6 +191,27 @@ mod tests {
     use crate::spec::stats::BinaryTableStats;
     use crate::spec::{DataFileMeta, FileKind, ManifestEntry};
 
+    #[tokio::test]
+    async fn test_query_auth_table_refuses_partition_listing() {
+        let table = crate::table::query_auth_table();
+        for (err, what) in [
+            (
+                table.partition_stats().await.unwrap_err(),
+                "partition stats",
+            ),
+            (
+                table.list_partitions().await.unwrap_err(),
+                "partition listing",
+            ),
+        ] {
+            assert!(
+                matches!(err, crate::Error::Unsupported { ref message }
+                    if message.contains("query-auth.enabled")),
+                "{what} on a query-auth.enabled table must fail closed, got {err:?}"
+            );
+        }
+    }
+
     /// Build a minimal synthetic ManifestEntry for unit testing.
     /// Mirrors the helper used in `spec::manifest` tests.
     fn make_entry(

crates/paimon/src/table/table_commit.rs

@@ -155,6 +155,8 @@ impl TableCommit {
         commit_identifier: i64,
         filter_committed: bool,
     ) -> Result<()> {
+        // A commit validates against the existing snapshot.
+        CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
         self.table.ensure_not_branch_reference_for_write()?;
 
         if commit_messages.is_empty() {
@@ -198,6 +200,8 @@ impl TableCommit {
         expected_snapshot_id: i64,
         commit_identifier: i64,
     ) -> Result<()> {
+        // A commit validates against the existing snapshot.
+        CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
         self.table.ensure_not_branch_reference_for_write()?;
 
         if commit_messages.is_empty() {
@@ -269,6 +273,8 @@ impl TableCommit {
         commit_identifier: i64,
         filter_committed: bool,
     ) -> Result<()> {
+        // A commit validates against the existing snapshot.
+        CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
         self.table.ensure_not_branch_reference_for_write()?;
 
         if commit_messages.is_empty() && static_partitions.is_none() {
@@ -518,6 +524,8 @@ impl TableCommit {
         commit_identifier: i64,
         filter_committed: bool,
     ) -> Result<()> {
+        // A commit validates against the existing snapshot.
+        CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
         self.table.ensure_not_branch_reference_for_write()?;
 
         if partitions.is_empty() {
@@ -596,6 +604,8 @@ impl TableCommit {
         commit_identifier: i64,
         filter_committed: bool,
     ) -> Result<()> {
+        // A commit validates against the existing snapshot.
+        CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
         self.table.ensure_not_branch_reference_for_write()?;
 
         self.try_commit(
@@ -2907,6 +2917,41 @@ mod tests {
     };
     use chrono::{DateTime, Utc};
 
+    #[tokio::test]
+    async fn test_query_auth_table_refuses_commit_paths() {
+        let table = crate::table::query_auth_table();
+        let commit = TableCommit::new(table.clone(), "test-user".to_string());
+
+        let refused = |err: crate::Error, what: &str| {
+            assert!(
+                matches!(err, crate::Error::Unsupported { ref message }
+                    if message.contains("query-auth.enabled")),
+                "{what} on a query-auth.enabled table must fail closed, got {err:?}"
+            );
+        };
+
+        refused(commit.commit(Vec::new()).await.unwrap_err(), "committing");
+        refused(
+            commit.overwrite(Vec::new(), None).await.unwrap_err(),
+            "overwriting",
+        );
+        refused(
+            commit.truncate_partitions(Vec::new()).await.unwrap_err(),
+            "truncating partitions",
+        );
+        refused(
+            commit.truncate_table().await.unwrap_err(),
+            "truncating the table",
+        );
+
+        refused(
+            crate::table::TableWrite::new(&table, "test-user".to_string())
+                .err()
+                .expect("opening a write must fail closed"),
+            "opening a write",
+        );
+    }
+
     fn test_file_io() -> FileIO {
         FileIOBuilder::new("memory").build().unwrap()
     }

crates/paimon/src/table/table_write.rs

@@ -130,6 +130,9 @@ pub struct TableWrite {
 
 impl TableWrite {
     pub(crate) fn new(table: &Table, commit_user: String) -> crate::Result<Self> {
+        // A dynamic-bucket write reads the persisted PK hash index; the rest are
+        // refused too, since their commit is blocked anyway.
+        CoreOptions::new(table.schema().options()).ensure_read_authorized()?;
         let is_overwrite = false;
         let schema = table.schema();
         let write_schema = build_target_arrow_schema(schema.fields())?;

crates/paimon/src/table/vindex_index_build_builder.rs

@@ -62,6 +62,9 @@ impl<'a> VindexIndexBuildBuilder<'a> {
     }
 
     pub async fn execute(&self) -> Result<usize> {
+        // Building the index scans the table's rows.
+        CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+
         self.table.ensure_not_branch_reference_for_write()?;
 
         if !is_vindex_index_type(&self.index_type) {