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/catalog/partition_listing.rs

@@ -33,7 +33,9 @@ use crate::Result;
 /// 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()?;
+    table
+        .ensure_read_authorized_live("listing partitions")
+        .await?;
     let file_io = table.file_io();
     let snapshot_sm = table.snapshot_manager();
     let manifest_sm = SnapshotManager::new(file_io.clone(), table.location().to_string());

crates/paimon/src/spec/schema.rs

@@ -638,22 +638,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/cow_writer.rs

@@ -207,7 +207,9 @@ impl CopyOnWriteMergeWriter {
     #[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()?;
+        self.table
+            .ensure_read_authorized_live("a copy-on-write rewrite")
+            .await?;
 
         if self.affected_files.is_empty() {
             return Ok(Vec::new());

crates/paimon/src/table/data_evolution_writer.rs

@@ -160,7 +160,9 @@ impl DataEvolutionWriter {
     #[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()?;
+        self.table
+            .ensure_read_authorized_live("a row-id update")
+            .await?;
 
         let total_matched: usize = self.matched_batches.iter().map(|b| b.num_rows()).sum();
         if total_matched == 0 {
@@ -478,7 +480,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()?;
+        self.table
+            .ensure_read_authorized_live("a row-id delete")
+            .await?;
 
         dedup_i64_in_place(&mut self.row_ids);
         if self.row_ids.is_empty() {

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/full_text_search_builder.rs

@@ -63,15 +63,24 @@ const FULL_TEXT_INDEX_SEARCH_CONCURRENCY: usize = 8;
 /// Reference: `org.apache.paimon.table.source.FullTextSearchBuilder`
 pub struct FullTextSearchBuilder<'a> {
     table: &'a Table,
+    /// Set when the caller already asked, so a delegated search does not repeat it.
+    authorized: bool,
     text_column: Option<String>,
     query_text: Option<String>,
     limit: Option<usize>,
     include_row_ids: Option<RoaringTreemap>,
 }
 
 impl<'a> FullTextSearchBuilder<'a> {
+    /// The caller already asked the server for this operation.
+    pub(crate) fn assume_authorized(mut self) -> Self {
+        self.authorized = true;
+        self
+    }
+
     pub(crate) fn new(table: &'a Table) -> Self {
         Self {
+            authorized: false,
             table,
             text_column: None,
             query_text: None,
@@ -117,7 +126,11 @@ impl<'a> FullTextSearchBuilder<'a> {
     pub async fn execute_scored(&self) -> crate::Result<SearchResult> {
         // Fail closed: returns data-derived row ranges outside `TableScan`/`TableRead`.
         let core = CoreOptions::new(self.table.schema().options());
-        core.ensure_read_authorized()?;
+        if !self.authorized {
+            self.table
+                .ensure_read_authorized_live("a full-text search")
+                .await?;
+        }
         let text_column =
             self.text_column
                 .as_deref()
@@ -204,7 +217,11 @@ impl<'a> FullTextSearchBuilder<'a> {
     pub async fn execute_read(&self) -> crate::Result<ArrowRecordBatchStream> {
         // Fail closed: returns data outside `TableScan`/`TableRead`.
         let core = CoreOptions::new(self.table.schema().options());
-        core.ensure_read_authorized()?;
+        if !self.authorized {
+            self.table
+                .ensure_read_authorized_live("a full-text search")
+                .await?;
+        }
         let text_column =
             self.text_column
                 .as_deref()

crates/paimon/src/table/global_index_drop_builder.rs

@@ -51,7 +51,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_read_authorized_live("dropping an index")
+            .await?;
 
         self.table.ensure_not_branch_reference_for_write()?;
 

crates/paimon/src/table/hybrid_search_builder.rs

@@ -287,7 +287,9 @@ impl<'a> HybridSearchBuilder<'a> {
 
     pub async fn execute_scored(&self) -> crate::Result<SearchResult> {
         let core = CoreOptions::new(self.table.schema().options());
-        core.ensure_read_authorized()?;
+        self.table
+            .ensure_read_authorized_live("a hybrid search")
+            .await?;
         let limit = self.limit.ok_or_else(|| crate::Error::ConfigInvalid {
             message: "Limit must be set via with_limit()".to_string(),
         })?;
@@ -318,7 +320,7 @@ impl<'a> HybridSearchBuilder<'a> {
         for route in &self.routes {
             let result = match route.kind {
                 HybridSearchRouteKind::Vector => {
-                    let mut builder = self.table.new_vector_search_builder();
+                    let mut builder = self.table.new_vector_search_builder().assume_authorized();
                     builder
                         .with_vector_column(&route.field_name)
                         .with_query_vector(route.vector.clone().expect("validated vector route"))
@@ -351,7 +353,9 @@ impl<'a> HybridSearchBuilder<'a> {
     /// `execute`/`execute_scored`. Mirrors Java `HybridSearchBuilderImpl` PK path.
     pub async fn execute_read(&self) -> crate::Result<ArrowRecordBatchStream> {
         let core = CoreOptions::new(self.table.schema().options());
-        core.ensure_read_authorized()?;
+        self.table
+            .ensure_read_authorized_live("a hybrid search")
+            .await?;
         let limit = self.limit.ok_or_else(|| crate::Error::ConfigInvalid {
             message: "Limit must be set via with_limit()".to_string(),
         })?;
@@ -572,7 +576,7 @@ impl<'a> HybridSearchBuilder<'a> {
         route: &HybridSearchRoute,
     ) -> crate::Result<PkRoute> {
         let vector = route.vector.as_deref().expect("validated vector route");
-        let mut builder = table.new_vector_search_builder();
+        let mut builder = table.new_vector_search_builder().assume_authorized();
         builder
             .with_vector_column(&route.field_name)
             .with_query_vector(vector.to_vec())
@@ -902,7 +906,7 @@ async fn execute_full_text_route(
     table: &Table,
     route: &HybridSearchRoute,
 ) -> crate::Result<SearchResult> {
-    let mut builder = table.new_full_text_search_builder();
+    let mut builder = table.new_full_text_search_builder().assume_authorized();
     builder
         .with_text_column(&route.field_name)
         .with_query_text(

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/lumina_index_build_builder.rs

@@ -70,7 +70,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_read_authorized_live("building an index")
+            .await?;
 
         self.table.ensure_not_branch_reference_for_write()?;