From 8019cbc0f89c6f6e05b53a321ac54f86ef267678 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Fri, 7 Aug 2026 07:51:35 -0400 Subject: [PATCH] fix(table): fail closed on query-auth tables in write, index and metadata paths --- .../paimon/src/catalog/partition_listing.rs | 18 +++++++ .../table/btree_global_index_build_builder.rs | 3 ++ .../paimon/src/table/bucket_assigner_cross.rs | 3 ++ crates/paimon/src/table/cow_writer.rs | 3 ++ .../paimon/src/table/data_evolution_writer.rs | 6 +++ .../src/table/global_index_drop_builder.rs | 50 +++++++++++++++++++ .../src/table/lumina_index_build_builder.rs | 3 ++ crates/paimon/src/table/partition_stat.rs | 23 +++++++++ crates/paimon/src/table/table_commit.rs | 45 +++++++++++++++++ crates/paimon/src/table/table_write.rs | 3 ++ .../src/table/vindex_index_build_builder.rs | 3 ++ 11 files changed, 160 insertions(+) diff --git a/crates/paimon/src/catalog/partition_listing.rs b/crates/paimon/src/catalog/partition_listing.rs index 3cb75041e..9bb28d309 100644 --- a/crates/paimon/src/catalog/partition_listing.rs +++ b/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> { + // 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 BTreeGlobalIndexBuildBuilder<'a> { } pub async fn execute(&self) -> Result { + // 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(|| { diff --git a/crates/paimon/src/table/bucket_assigner_cross.rs b/crates/paimon/src/table/bucket_assigner_cross.rs index f0e711b2a..5a1d33988 100644 --- a/crates/paimon/src/table/bucket_assigner_cross.rs +++ b/crates/paimon/src/table/bucket_assigner_cross.rs @@ -71,6 +71,9 @@ impl GlobalPartitionIndex { target_bucket_row_number: i64, merge_engine: MergeEngine, ) -> Result { + // 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, i32)> = HashMap::new(); let mut bucket_row_counts: HashMap<(Vec, i32), i64> = HashMap::new(); diff --git a/crates/paimon/src/table/cow_writer.rs b/crates/paimon/src/table/cow_writer.rs index 6c348d15c..1f77c8721 100644 --- a/crates/paimon/src/table/cow_writer.rs +++ b/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> { + // 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()); } diff --git a/crates/paimon/src/table/data_evolution_writer.rs b/crates/paimon/src/table/data_evolution_writer.rs index 5e16524f6..078a2a128 100644 --- a/crates/paimon/src/table/data_evolution_writer.rs +++ b/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> { + // 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> { + // 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()); diff --git a/crates/paimon/src/table/global_index_drop_builder.rs b/crates/paimon/src/table/global_index_drop_builder.rs index cc86ceb07..80a0ad7f8 100644 --- a/crates/paimon/src/table/global_index_drop_builder.rs +++ b/crates/paimon/src/table/global_index_drop_builder.rs @@ -50,6 +50,9 @@ impl<'a> GlobalIndexDropBuilder<'a> { } pub async fn execute(&self) -> Result { + // 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())) diff --git a/crates/paimon/src/table/lumina_index_build_builder.rs b/crates/paimon/src/table/lumina_index_build_builder.rs index 5e45abf8e..b64a8e90f 100644 --- a/crates/paimon/src/table/lumina_index_build_builder.rs +++ b/crates/paimon/src/table/lumina_index_build_builder.rs @@ -71,6 +71,9 @@ impl<'a> LuminaIndexBuildBuilder<'a> { } pub async fn execute(&self) -> Result { + // 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) { diff --git a/crates/paimon/src/table/partition_stat.rs b/crates/paimon/src/table/partition_stat.rs index 9d78b58ad..f24719713 100644 --- a/crates/paimon/src/table/partition_stat.rs +++ b/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> { + // 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( diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index 809e45d9f..256b3a33a 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/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() } diff --git a/crates/paimon/src/table/table_write.rs b/crates/paimon/src/table/table_write.rs index bb152a67f..087f248af 100644 --- a/crates/paimon/src/table/table_write.rs +++ b/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 { + // 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())?; diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs b/crates/paimon/src/table/vindex_index_build_builder.rs index 11579c6eb..7bc889223 100644 --- a/crates/paimon/src/table/vindex_index_build_builder.rs +++ b/crates/paimon/src/table/vindex_index_build_builder.rs @@ -62,6 +62,9 @@ impl<'a> VindexIndexBuildBuilder<'a> { } pub async fn execute(&self) -> Result { + // 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) {