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 992aced5 fix(auth): fail closed on query-auth tables outside the read
boundary (#691)
992aced5 is described below
commit 992aced55f596d058d642a1695f875da8655c7a4
Author: Jiajia Li <[email protected]>
AuthorDate: Sat Aug 15 16:14:08 2026 +0800
fix(auth): fail closed on query-auth tables outside the read boundary (#691)
---
crates/paimon/src/catalog/partition_listing.rs | 18 ++++++++
.../src/table/btree_global_index_build_builder.rs | 3 ++
crates/paimon/src/table/bucket_assigner_cross.rs | 3 ++
crates/paimon/src/table/cow_writer.rs | 3 ++
crates/paimon/src/table/data_evolution_writer.rs | 6 +++
.../paimon/src/table/global_index_drop_builder.rs | 50 ++++++++++++++++++++++
.../paimon/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 ++
.../paimon/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 3cb75041..9bb28d30 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<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:?}"
+ );
+ }
+}
diff --git a/crates/paimon/src/table/btree_global_index_build_builder.rs
b/crates/paimon/src/table/btree_global_index_build_builder.rs
index 4692a62c..608b0579 100644
--- a/crates/paimon/src/table/btree_global_index_build_builder.rs
+++ b/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(|| {
diff --git a/crates/paimon/src/table/bucket_assigner_cross.rs
b/crates/paimon/src/table/bucket_assigner_cross.rs
index f0e711b2..5a1d3398 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<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();
diff --git a/crates/paimon/src/table/cow_writer.rs
b/crates/paimon/src/table/cow_writer.rs
index 6c348d15..1f77c872 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<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());
}
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index 5e16524f..078a2a12 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<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());
diff --git a/crates/paimon/src/table/global_index_drop_builder.rs
b/crates/paimon/src/table/global_index_drop_builder.rs
index cc86ceb0..80a0ad7f 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<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()))
diff --git a/crates/paimon/src/table/lumina_index_build_builder.rs
b/crates/paimon/src/table/lumina_index_build_builder.rs
index 5e45abf8..b64a8e90 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<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) {
diff --git a/crates/paimon/src/table/partition_stat.rs
b/crates/paimon/src/table/partition_stat.rs
index 9d78b58a..ae5e5580 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<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_stats_and_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 96fb45f3..2ce701fd 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() {
@@ -279,6 +283,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() {
@@ -528,6 +534,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() {
@@ -606,6 +614,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(
@@ -2917,6 +2927,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 bb152a67..087f248a 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<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())?;
diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs
b/crates/paimon/src/table/vindex_index_build_builder.rs
index bde7d0ab..7638f28c 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -66,6 +66,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) {
