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

crates/paimon/src/api/api_response.rs

@@ -476,8 +476,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>>,
@@ -495,6 +498,18 @@ impl AuthTableQueryResponse {
 
 #[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]

crates/paimon/src/spec/schema.rs

@@ -621,22 +621,27 @@ impl TableSchema {
     }
 }
 
-/// Reject column names reserved for system use, mirroring Java `SpecialFields`:
-/// the five `SYSTEM_FIELD_NAMES` and the `_KEY_` key-field prefix.
+/// Whether `name` is one Paimon reserves for a system column. Java
+/// `SpecialFields.SYSTEM_FIELD_NAMES` plus the `_KEY_` key-field prefix.
+pub(crate) fn is_reserved_system_field_name(name: &str) -> bool {
+    name.starts_with(KEY_FIELD_PREFIX) || SYSTEM_FIELD_NAMES.contains(&name)
+}
+
+// Java SpecialFields.SYSTEM_FIELD_NAMES.
+const SYSTEM_FIELD_NAMES: [&str; 5] = [
+    SEQUENCE_NUMBER_FIELD_NAME,
+    VALUE_KIND_FIELD_NAME,
+    "_LEVEL",
+    ROW_KIND_FIELD_NAME,
+    ROW_ID_FIELD_NAME,
+];
+const KEY_FIELD_PREFIX: &str = "_KEY_";
+
+/// Reject column names reserved for system use, mirroring Java `SpecialFields`.
 ///
 /// A user column colliding with a system field is otherwise excluded from the
 /// physical read and silently filled with the system value.
 fn validate_no_reserved_field_names(fields: &[DataField]) -> crate::Result<()> {
-    // Java SpecialFields.SYSTEM_FIELD_NAMES.
-    const SYSTEM_FIELD_NAMES: [&str; 5] = [
-        SEQUENCE_NUMBER_FIELD_NAME,
-        VALUE_KIND_FIELD_NAME,
-        "_LEVEL",
-        ROW_KIND_FIELD_NAME,
-        ROW_ID_FIELD_NAME,
-    ];
-    const KEY_FIELD_PREFIX: &str = "_KEY_";
-
     for field in fields {
         let name = field.name();
         if name.starts_with(KEY_FIELD_PREFIX) || SYSTEM_FIELD_NAMES.contains(&name) {

crates/paimon/src/table/format_table_read.rs

@@ -103,7 +103,15 @@ impl<'a> FormatTableRead<'a> {
         data_splits: &[DataSplit],
     ) -> crate::Result<ArrowRecordBatchStream> {
         let core_options = self.table.schema().core_options();
-        core_options.ensure_read_authorized()?;
+        core_options.ensure_type_paimon_served(&self.table.identifier().full_name())?;
+        // Sync, so the marker stands in for asking the server.
+        if core_options.query_auth_enabled()
+            || data_splits.iter().any(|split| split.query_auth_required())
+        {
+            return Err(super::query_auth::unsupported(
+                "a format table cannot apply a row filter or column masking",
+            ));
+        }
         // Mapping the conjunct onto the data fields drops it, so the read would
         // silently ignore the filter. Guard on the read path, not the builder:
         // `TableRead` is public and can be constructed and filtered directly.

crates/paimon/src/table/format_table_scan.rs

@@ -56,20 +56,28 @@ impl<'a> FormatTableScan<'a> {
     }
 
     pub(crate) async fn plan(&self) -> crate::Result<Plan> {
-        self.ensure_query_auth_allowed()?;
+        self.ensure_query_auth_allowed().await?;
         self.plan_inner(None).await
     }
 
     pub(crate) async fn plan_with_trace(&self) -> crate::Result<(Plan, ScanTrace)> {
-        self.ensure_query_auth_allowed()?;
+        self.ensure_query_auth_allowed().await?;
         let mut trace = ScanTrace::default();
         let plan = self.plan_inner(Some(&mut trace)).await?;
         trace.planned_data_file_bytes = plan.planned_data_file_bytes();
         Ok((plan, trace))
     }
 
-    fn ensure_query_auth_allowed(&self) -> crate::Result<()> {
-        CoreOptions::new(self.table.schema().options()).ensure_read_authorized()
+    /// Refused outright. Asks the server: the option can be set after a load.
+    async fn ensure_query_auth_allowed(&self) -> crate::Result<()> {
+        let core_options = CoreOptions::new(self.table.schema().options());
+        core_options.ensure_type_paimon_served(&self.table.identifier().full_name())?;
+        if self.table.server_query_auth_enabled().await? {
+            return Err(super::query_auth::unsupported(
+                "a format table cannot apply a row filter or column masking",
+            ));
+        }
+        Ok(())
     }
 
     async fn plan_inner(&self, trace: Option<&mut ScanTrace>) -> crate::Result<Plan> {

crates/paimon/src/table/incremental_scan.rs

@@ -149,6 +149,18 @@ impl IncrementalPlan {
         &self.splits
     }
 
+    /// Whether any underlying split came from a query-auth plan. Unlike
+    /// [`Self::data_splits`] this sees the diff pairs too.
+    pub(crate) fn any_query_auth_required(&self) -> bool {
+        self.splits.iter().any(|split| match split {
+            IncrementalSplit::Data(split) => split.query_auth_required(),
+            IncrementalSplit::DiffPair { before, after } => before
+                .iter()
+                .chain(after)
+                .any(DataSplit::query_auth_required),
+        })
+    }
+
     pub fn data_splits(&self) -> Vec<DataSplit> {
         self.splits
             .iter()
@@ -244,7 +256,13 @@ impl<'a> IncrementalScan<'a> {
     }
 
     pub async fn plan(&self) -> crate::Result<IncrementalPlan> {
-        crate::spec::CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+        let core_options = crate::spec::CoreOptions::new(self.table.schema().options());
+        core_options.ensure_type_paimon_served(&self.table.identifier().full_name())?;
+        if self.table.server_query_auth_enabled().await? {
+            return Err(super::query_auth::unsupported(
+                "an incremental read cannot apply a row filter or column masking",
+            ));
+        }
         let mode = self.resolve_mode();
         self.validate_snapshot_range(mode).await?;
         if self.start_exclusive == self.end_inclusive {

crates/paimon/src/table/mod.rs

@@ -81,6 +81,7 @@ mod postpone_fixed_bucket_router;
 mod postpone_fixed_bucket_write;
 mod postpone_fixed_bucket_write_builder;
 mod prepared_files;
+mod query_auth;
 mod read_builder;
 pub mod referenced_files;
 pub(crate) mod rest_env;
@@ -172,6 +173,9 @@ pub struct Table {
     schema_manager: SchemaManager,
     branch: String,
     branch_reference: bool,
+    /// Minted only by [`RESTEnv::build_table`], so a handle assembled with the
+    /// public [`Table::new`] cannot replay a grant.
+    query_auth_session: Option<u64>,
     rest_env: Option<RESTEnv>,
     /// True when this table copy was switched to a historical schema by
     /// [`Table::copy_with_time_travel`]. Such a copy is read-only.
@@ -201,6 +205,7 @@ impl Table {
             schema_manager,
             branch,
             branch_reference: false,
+            query_auth_session: None,
             rest_env,
             time_traveled: false,
             travel_snapshot: None,
@@ -241,6 +246,7 @@ impl Table {
             schema_manager,
             branch,
             branch_reference,
+            query_auth_session: None,
             rest_env: None,
             time_traveled: false,
             travel_snapshot: None,
@@ -315,6 +321,97 @@ impl Table {
         }
     }
 
+    /// Whether the server says this table is `query-auth.enabled` right now: the
+    /// handle's schema is a snapshot, and a cached `false` would skip the check.
+    pub(crate) async fn server_query_auth_enabled(&self) -> Result<bool> {
+        let local = CoreOptions::new(self.schema.options()).query_auth_enabled();
+        let Some(rest_env) = &self.rest_env else {
+            return Ok(local);
+        };
+        // Only ever strengthens: the name can be re-created over this handle's
+        // files, so the answer may be about a different table.
+        if local {
+            return Ok(true);
+        }
+        match rest_env.current_table().await?.schema.as_ref() {
+            Some(schema) => Ok(CoreOptions::new(schema.options()).query_auth_enabled()),
+            None => Ok(true),
+        }
+    }
+
+    /// Whether this user may read this table; `None` when it is not
+    /// `query-auth.enabled`. The caller stamps the grant onto its splits.
+    ///
+    /// `server_query_auth` is the caller's already-fetched
+    /// [`Self::server_query_auth_enabled`], so planning asks the server once.
+    pub(crate) async fn authorize_read(
+        &self,
+        server_query_auth: bool,
+    ) -> Result<Option<std::sync::Arc<query_auth::QueryAuthGrant>>> {
+        let local = CoreOptions::new(self.schema.options());
+        // Ask the selector too: `copy_with_options` adds one without the flag.
+        let travels = local.try_time_travel_selector()?.is_some();
+        // A `$branch_x` or `$files` handle authorizes against the decorated
+        // name while its managers read the base table's own files.
+        let decorated = self.identifier.branch_name()?.is_some()
+            || self.identifier.system_table_name()?.is_some();
+        if (travels || self.time_traveled || self.branch_reference || decorated)
+            && local.query_auth_enabled()
+        {
+            return Err(query_auth::unsupported(
+                "a time-travelled or branch read authorizes against the table's current schema, \
+                 which is not the one it reads",
+            ));
+        }
+
+        let Some(rest_env) = &self.rest_env else {
+            // Only a REST catalog can authorize.
+            return if local.query_auth_enabled() {
+                Err(query_auth::unsupported(
+                    "it requires a REST catalog to authorize the query",
+                ))
+            } else {
+                Ok(None)
+            };
+        };
+
+        // No freshness assertion yet — an ordinary table must not inherit one.
+        if !server_query_auth {
+            return Ok(None);
+        }
+        if travels || self.time_traveled || self.branch_reference || decorated {
+            return Err(query_auth::unsupported(
+                "a time-travelled or branch read authorizes against the table's current schema, \
+                 which is not the one it reads",
+            ));
+        }
+
+        // Before any RPC: only the catalog mints a session, so a handle the
+        // caller assembled stops here whatever name or files it wears.
+        let session = self.query_auth_session.ok_or_else(|| {
+            query_auth::unsupported("this table handle was assembled rather than loaded")
+        })?;
+
+        // Naming a system column here would fail the server's column check.
+        let response = rest_env
+            .table_query_auth(&self.branch, self.schema.id(), None)
+            .await?;
+        Ok(Some(std::sync::Arc::new(query_auth::QueryAuthGrant::new(
+            response, session,
+        ))))
+    }
+
+    /// Handed out once per catalog-loaded table; wraps only after 2^64 loads.
+    pub(crate) fn with_query_auth_session(mut self) -> Self {
+        static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
+        self.query_auth_session = Some(NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed));
+        self
+    }
+
+    pub(crate) fn query_auth_session(&self) -> Option<u64> {
+        self.query_auth_session
+    }
+
     /// Get the REST environment, if this table was loaded from a REST catalog.
     pub fn rest_env(&self) -> Option<&RESTEnv> {
         self.rest_env.as_ref()
@@ -414,6 +511,7 @@ impl Table {
             schema_manager: self.schema_manager.clone(),
             branch: self.branch.clone(),
             branch_reference: self.branch_reference,
+            query_auth_session: self.query_auth_session,
             rest_env: self.rest_env.clone(),
             time_traveled: self.time_traveled,
             travel_snapshot: if selector_changed {
@@ -524,6 +622,7 @@ impl Table {
             schema_manager,
             branch,
             branch_reference: true,
+            query_auth_session: self.query_auth_session,
             rest_env: self.rest_env.clone(),
             time_traveled: false,
             travel_snapshot: None,
@@ -558,6 +657,32 @@ pub(crate) fn find_field_id_by_name(fields: &[DataField], name: &str) -> Option<
     fields.iter().find(|f| f.name() == name).map(|f| f.id())
 }
 
+/// A `query-auth.enabled` table wired to its own REST session.
+#[cfg(test)]
+pub(crate) async fn rest_query_auth_table() -> Table {
+    use crate::api::rest_api::RESTApi;
+    use crate::common::{CatalogOptions, Options};
+
+    let mut options = Options::default();
+    options.set(CatalogOptions::URI, "http://127.0.0.1:1");
+    options.set("token.provider", "bear");
+    options.set("token", "test_token");
+    let api = std::sync::Arc::new(RESTApi::new(options.clone(), false).await.unwrap());
+    let table = query_auth_table();
+    Table {
+        rest_env: Some(RESTEnv::new(
+            table.identifier.clone(),
+            "uuid-1".to_string(),
+            api,
+            options,
+            false,
+            None,
+        )),
+        ..table
+    }
+    .with_query_auth_session()
+}
+
 /// A minimal table with `query-auth.enabled = true`, for the fail-closed read guard.
 #[cfg(test)]
 pub(crate) fn query_auth_table() -> Table {