feat(auth): authorize query-auth reads and carry the grant on the split

crates/integrations/datafusion/src/system_tables/branches.rs

@@ -74,6 +74,7 @@ impl TableProvider for BranchesTable {
         _filters: &[Expr],
         _limit: Option<usize>,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        super::ensure_scan_authorized(&self.table).await?;
         let table = self.table.clone();
         let (names, create_times) =
             crate::runtime::await_with_runtime(async move { collect_branches(&table).await })

crates/integrations/datafusion/src/system_tables/consumers.rs

@@ -72,6 +72,7 @@ impl TableProvider for ConsumersTable {
         filters: &[Expr],
         _limit: Option<usize>,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        super::ensure_scan_authorized(&self.table).await?;
         let manager = self.table.consumer_manager();
         let requested_ids = requested_consumer_ids(filters);
         let consumers = crate::runtime::await_with_runtime(async move {

crates/integrations/datafusion/src/system_tables/files.rs

@@ -105,6 +105,7 @@ impl TableProvider for FilesTable {
         _filters: &[Expr],
         _limit: Option<usize>,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        super::ensure_scan_authorized(&self.table).await?;
         let table = self.table.clone();
         let rows =
             crate::runtime::await_with_runtime(async move { collect_file_rows(&table).await })

crates/integrations/datafusion/src/system_tables/manifests.rs

@@ -82,6 +82,7 @@ impl TableProvider for ManifestsTable {
         _filters: &[Expr],
         _limit: Option<usize>,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        super::ensure_scan_authorized(&self.table).await?;
         let table = self.table.clone();
         let metas =
             crate::runtime::await_with_runtime(async move { collect_manifests(&table).await })

crates/integrations/datafusion/src/system_tables/mod.rs

@@ -124,6 +124,14 @@ fn wrap_to_system_table(name: &str, base_table: Table) -> Option<DFResult<Arc<dy
         .map(|(_, build)| build(base_table))
 }
 
+/// Fail closed at scan time, asking the server: the option can be set after
+/// the provider was built.
+pub(crate) async fn ensure_scan_authorized(table: &Table) -> DFResult<()> {
+    crate::runtime::await_with_runtime(table.ensure_read_authorized())
+        .await
+        .map_err(to_datafusion_error)
+}
+
 pub(crate) fn provider_for_table(
     catalog: Arc<dyn Catalog>,
     identifier: Identifier,

crates/integrations/datafusion/src/system_tables/options.rs

@@ -68,6 +68,7 @@ impl TableProvider for OptionsTable {
         _filters: &[Expr],
         _limit: Option<usize>,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        super::ensure_scan_authorized(&self.table).await?;
         // Java uses LinkedHashMap insertion order; HashMap has none — sort for stable output.
         let mut entries: Vec<(&String, &String)> = self.table.schema().options().iter().collect();
         entries.sort_by(|a, b| a.0.cmp(b.0));

crates/integrations/datafusion/src/system_tables/partitions.rs

@@ -121,6 +121,7 @@ impl TableProvider for PartitionsTable {
         _filters: &[Expr],
         _limit: Option<usize>,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        super::ensure_scan_authorized(&self.table).await?;
         let table = self.table.clone();
         let partitions = if table.travel_snapshot().is_some() {
             crate::runtime::await_with_runtime(async move {

crates/integrations/datafusion/src/system_tables/physical_files_size.rs

@@ -75,6 +75,7 @@ impl TableProvider for PhysicalFilesSizeTable {
         _filters: &[Expr],
         _limit: Option<usize>,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        super::ensure_scan_authorized(&self.table).await?;
         let table = self.table.clone();
         let summary = crate::runtime::await_with_runtime(async move {
             let partition_depth = table.schema().partition_keys().len();

crates/integrations/datafusion/src/system_tables/referenced_files_size.rs

@@ -76,6 +76,7 @@ impl TableProvider for ReferencedFilesSizeTable {
         _filters: &[Expr],
         _limit: Option<usize>,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        super::ensure_scan_authorized(&self.table).await?;
         let table = self.table.clone();
         let summaries = crate::runtime::await_with_runtime(async move {
             let schema = table.schema();

crates/integrations/datafusion/src/system_tables/schemas.rs

@@ -80,6 +80,7 @@ impl TableProvider for SchemasTable {
         _filters: &[Expr],
         _limit: Option<usize>,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        super::ensure_scan_authorized(&self.table).await?;
         let table = self.table.clone();
         let schemas =
             crate::runtime::await_with_runtime(

crates/integrations/datafusion/src/system_tables/snapshots.rs

@@ -86,6 +86,7 @@ impl TableProvider for SnapshotsTable {
         _filters: &[Expr],
         _limit: Option<usize>,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        super::ensure_scan_authorized(&self.table).await?;
         let sm = self.table.snapshot_manager();
         let snapshots = crate::runtime::await_with_runtime(async move { sm.list_all().await })
             .await

crates/integrations/datafusion/src/system_tables/table_indexes.rs

@@ -104,6 +104,7 @@ impl TableProvider for TableIndexesTable {
         _filters: &[Expr],
         _limit: Option<usize>,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        super::ensure_scan_authorized(&self.table).await?;
         let table = self.table.clone();
         let entries =
             crate::runtime::await_with_runtime(async move { collect_index_entries(&table).await })

crates/integrations/datafusion/src/system_tables/tags.rs

@@ -83,6 +83,7 @@ impl TableProvider for TagsTable {
         _filters: &[Expr],
         _limit: Option<usize>,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        super::ensure_scan_authorized(&self.table).await?;
         let tm = self.table.tag_manager();
         let tags =
             crate::runtime::await_with_runtime(async move { tm.list_all_with_metadata().await })

crates/paimon-rest-server/src/lib.rs

@@ -437,10 +437,11 @@ async fn get_table(path: RestPath, Extension(state): Extension<Arc<AppState>>) -
             }
         };
 
+    // FileSystemCatalog has no UUID concept; the full name is a stable id that
+    // satisfies the client's RESTEnv requirement.
+    let uuid = identifier.full_name();
     let response = GetTableResponse::new(
-        // FileSystemCatalog has no UUID concept; the full name is a stable id
-        // that satisfies the client's RESTEnv requirement.
-        Some(identifier.full_name()),
+        Some(uuid),
         Some(table),
         Some(location),
         Some(false),

crates/paimon-rest-server/tests/e2e.rs

@@ -525,3 +525,173 @@ async fn altering_the_declared_type_is_rejected() {
         .await
         .expect("still readable");
 }
+
+#[tokio::test]
+async fn test_branch_scan_against_the_real_server() {
+    let ctx = setup().await;
+    ctx.catalog
+        .create_database("db", true, HashMap::new())
+        .await
+        .unwrap();
+    let identifier = Identifier::new("db", "t");
+    ctx.catalog
+        .create_table(&identifier, append_only_schema(), false)
+        .await
+        .unwrap();
+    let base = ctx.catalog.get_table(&identifier).await.unwrap();
+
+    // A branch schema on disk, so `copy_with_branch` and the server both see it.
+    let branch_schema = paimon::spec::TableSchema::new(0, &append_only_schema());
+    let schema_path = base.schema_manager().with_branch("dev").schema_path(0);
+    let schema_dir = schema_path.rsplit_once('/').map(|(d, _)| d).unwrap();
+    base.file_io().mkdirs(schema_dir).await.unwrap();
+    base.file_io()
+        .new_output(&schema_path)
+        .unwrap()
+        .write(serde_json::to_vec(&branch_schema).unwrap().into())
+        .await
+        .unwrap();
+
+    // The branch reports the base table's uuid, so an ordinary branch scan
+    // through the copied handle still plans.
+    base.copy_with_branch("dev")
+        .await
+        .unwrap()
+        .new_read_builder()
+        .new_scan()
+        .plan()
+        .await
+        .expect("an ordinary branch read must plan against the real server");
+
+    // A decorated name is answered by the server for the live check only;
+    // the catalog never builds a handle from one.
+    assert!(ctx
+        .catalog
+        .get_table(&Identifier::new("db", "t$branch_dev"))
+        .await
+        .is_err());
+}
+
+#[tokio::test]
+async fn test_a_commit_addressed_to_a_branch_is_refused() {
+    let ctx = setup().await;
+    ctx.catalog
+        .create_database("db", true, HashMap::new())
+        .await
+        .unwrap();
+    let identifier = Identifier::new("db", "t");
+    ctx.catalog
+        .create_table(&identifier, append_only_schema(), false)
+        .await
+        .unwrap();
+    let base = ctx.catalog.get_table(&identifier).await.unwrap();
+    let schema_path = base.schema_manager().with_branch("dev").schema_path(0);
+    let schema_dir = schema_path.rsplit_once('/').map(|(d, _)| d).unwrap();
+    base.file_io().mkdirs(schema_dir).await.unwrap();
+    base.file_io()
+        .new_output(&schema_path)
+        .unwrap()
+        .write(
+            serde_json::to_vec(&paimon::spec::TableSchema::new(0, &append_only_schema()))
+                .unwrap()
+                .into(),
+        )
+        .await
+        .unwrap();
+
+    // Straight at the endpoint, past the client's own branch-write refusal:
+    // the server used to resolve the branch and then commit to main.
+    let snapshot = paimon::spec::Snapshot::builder()
+        .version(3)
+        .id(1)
+        .schema_id(0)
+        .base_manifest_list("manifest-list-0".to_string())
+        .delta_manifest_list("manifest-list-1".to_string())
+        .commit_user("e2e".to_string())
+        .commit_identifier(1)
+        .commit_kind(paimon::spec::CommitKind::APPEND)
+        .time_millis(0)
+        .build();
+    let outcome = base
+        .rest_env()
+        .unwrap()
+        .api()
+        .commit_snapshot(
+            &Identifier::new("db", "t$branch_dev"),
+            "db.t",
+            &snapshot,
+            &[],
+        )
+        .await;
+    assert!(
+        outcome.is_err(),
+        "a commit addressed to a branch must be refused"
+    );
+    assert!(
+        base.snapshot_manager()
+            .get_latest_snapshot_id()
+            .await
+            .unwrap()
+            .is_none(),
+        "and main must be untouched"
+    );
+}
+
+#[tokio::test]
+async fn test_load_table_refuses_a_decorated_object_table() {
+    let ctx = setup().await;
+    ctx.catalog
+        .create_database("db", true, HashMap::new())
+        .await
+        .unwrap();
+    let identifier = Identifier::new("db", "objects");
+    let schema = paimon::spec::Schema::builder()
+        .column(
+            "ignored",
+            paimon::spec::DataType::Int(paimon::spec::IntType::new()),
+        )
+        .option("type", "object-table")
+        .build()
+        .unwrap();
+    ctx.catalog
+        .create_table(&identifier, schema, false)
+        .await
+        .unwrap();
+    // With a branch schema on disk the server resolves the name, so only the
+    // client's own refusal keeps `load_table`'s object-table early return from
+    // handing back the base relation.
+    let loaded = ctx.catalog.load_table(&identifier).await.unwrap();
+    let paimon::catalog::LoadedTable::Object(object) = loaded else {
+        panic!("expected an object table");
+    };
+    let manager =
+        paimon::table::SchemaManager::new(object.file_io().clone(), object.location().to_string())
+            .with_branch("dev");
+    let schema_path = manager.schema_path(0);
+    let schema_dir = schema_path.rsplit_once('/').map(|(d, _)| d).unwrap();
+    object.file_io().mkdirs(schema_dir).await.unwrap();
+    let (_, stored) = paimon::catalog::FileSystemCatalog::new({
+        let mut o = Options::new();
+        o.set(
+            CatalogOptions::WAREHOUSE,
+            ctx._warehouse.path().to_str().unwrap(),
+        );
+        o
+    })
+    .unwrap()
+    .fetch_table_schema(&identifier)
+    .await
+    .unwrap();
+    object
+        .file_io()
+        .new_output(&schema_path)
+        .unwrap()
+        .write(serde_json::to_vec(&stored).unwrap().into())
+        .await
+        .unwrap();
+    assert!(ctx
+        .catalog
+        .load_table(&Identifier::new("db", "objects$branch_dev"))
+        .await
+        .is_err());
+}

crates/paimon/src/api/api_response.rs

@@ -487,8 +487,11 @@ pub struct GetTableTokenResponse {
 
 /// Response for auth table query: the per-user row filter and column masking the
 /// client must enforce at read time for a `query-auth.enabled` table.
+///
+/// Unknown fields are rejected: an absent one reads as "no rule", so protocol
+/// drift would look like an unrestricted grant.
 #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct AuthTableQueryResponse {
     /// JSON-serialized row-filter predicates, ANDed together. Empty/None = no filter.
     pub filter: Option<Vec<String>>,
@@ -525,6 +528,18 @@ impl ListPermissionsResponse {
 
 #[cfg(test)]
 mod tests {
+
+    #[test]
+    fn test_auth_table_query_response_rejects_unknown_fields() {
+        let drifted = r#"{"rowFilter":["restricted"]}"#;
+        assert!(
+            serde_json::from_str::<AuthTableQueryResponse>(drifted).is_err(),
+            "an auth response this client does not understand must not parse"
+        );
+        assert!(serde_json::from_str::<AuthTableQueryResponse>("{}")
+            .unwrap()
+            .is_unrestricted());
+    }
     use super::*;
 
     #[test]