[table] Support query-auth row filters and column masking at read time

crates/integrations/datafusion/Cargo.toml

@@ -50,6 +50,8 @@ uuid = { version = "1", features = ["v4"] }
 [dev-dependencies]
 arrow-array = { workspace = true }
 arrow-schema = { workspace = true }
+# for the shared REST catalog mock (crates/paimon/tests/mock_server.rs)
+axum = { version = "0.7", features = ["macros", "tokio", "http1", "http2"] }
 bytes = "1.7.1"
 flate2 = "1"
 paimon-ftindex-core = "0.1.0"

crates/integrations/datafusion/src/merge_into.rs

@@ -1233,12 +1233,13 @@ pub(crate) async fn register_cow_target_table(
         return Ok((false, table_name));
     }
 
-    // Read all files in parallel
-    let read_futures: Vec<_> = file_index
-        .iter()
-        .enumerate()
-        .map(|(file_idx, file_info)| async move {
-            let single_split = DataSplitBuilder::new()
+    // This read rewrites the target files, so it must see raw rows. The scan
+    // plan's grant would shift the positional row offsets the writer replays,
+    // rewriting the wrong rows.
+    let mut splits = Vec::with_capacity(file_index.len());
+    for file_info in file_index.iter() {
+        splits.push(
+            DataSplitBuilder::new()
                 .with_snapshot(file_info.snapshot_id)
                 .with_partition(
                     paimon::spec::BinaryRow::from_serialized_bytes(&file_info.partition)
@@ -1249,8 +1250,19 @@ pub(crate) async fn register_cow_target_table(
                 .with_total_buckets(file_info.total_buckets)
                 .with_data_files(vec![file_info.file_meta.clone()])
                 .build()
-                .map_err(to_datafusion_error)?;
+                .map_err(to_datafusion_error)?,
+        );
+    }
+    let splits = table
+        .authorize_rewrite_splits(splits)
+        .await
+        .map_err(to_datafusion_error)?;
 
+    // Read all files in parallel
+    let read_futures: Vec<_> = splits
+        .into_iter()
+        .enumerate()
+        .map(|(file_idx, single_split)| async move {
             let read = table
                 .new_read_builder()
                 .new_read()

crates/integrations/datafusion/src/physical_plan/scan.rs

@@ -768,6 +768,10 @@ pub struct PaimonTableScan {
     /// Column-name case sensitivity carried from planning to execution so the
     /// read path resolves names the same way the scan was planned.
     case_sensitive: bool,
+    /// Set from [`paimon::table::Plan::planned_under_restricted_grant`]. Kept
+    /// on the plan rather than inferred from the splits, so a fully pruned scan
+    /// still suppresses its metadata.
+    query_auth_restricted: bool,
     /// Physical filters retained from DataFusion's runtime filter-pushdown pass.
     /// They are evaluated exactly by this scan.
     runtime_filters: Vec<Arc<dyn PhysicalExpr>>,
@@ -778,6 +782,22 @@ pub struct PaimonTableScan {
 }
 
 impl PaimonTableScan {
+    /// Record that this scan was planned under a row filter or column masking.
+    pub(crate) fn with_query_auth_restricted(mut self, restricted: bool) -> Self {
+        self.query_auth_restricted = restricted;
+        self
+    }
+
+    /// Whether this scan runs under a row filter or column masking.
+    fn is_query_auth_restricted(&self) -> bool {
+        self.query_auth_restricted
+            || self
+                .planned_partitions
+                .iter()
+                .flat_map(|splits| splits.iter())
+                .any(paimon::DataSplit::has_restricted_query_auth_grant)
+    }
+
     #[allow(clippy::too_many_arguments)]
     pub(crate) fn new(
         schema: ArrowSchemaRef,
@@ -808,6 +828,7 @@ impl PaimonTableScan {
             scan_trace,
             pushed_variants,
             case_sensitive,
+            query_auth_restricted: false,
             runtime_filters: Vec::new(),
             decoder_filters: Vec::new(),
         }
@@ -851,6 +872,16 @@ impl PaimonTableScan {
             return Statistics::unknown_column(&self.schema());
         }
 
+        // Manifest stats precede enforcement: a masked column's bounds are the
+        // raw values it hides, and null counts cover filtered-out rows.
+        if partitions
+            .iter()
+            .flat_map(|splits| splits.iter())
+            .any(DataSplit::has_restricted_query_auth_grant)
+        {
+            return Statistics::unknown_column(&self.schema());
+        }
+
         let Ok(merge_engine) = self.table.schema().core_options().merge_engine() else {
             return Statistics::unknown_column(&self.schema());
         };
@@ -1066,6 +1097,16 @@ impl ExecutionPlan for PaimonTableScan {
             None => &self.planned_partitions,
         };
 
+        // The manifest row count precedes the filter, so it would disclose how
+        // much data is hidden. Report the whole thing as unknown.
+        if partitions
+            .iter()
+            .flat_map(|splits| splits.iter())
+            .any(DataSplit::has_restricted_query_auth_grant)
+        {
+            return Ok(Arc::new(Statistics::new_unknown(&self.schema())));
+        }
+
         let mut total_rows: usize = 0;
         let mut all_row_counts_known = true;
         for splits in partitions {
@@ -1109,6 +1150,19 @@ impl DisplayAs for PaimonTableScan {
     ) -> std::fmt::Result {
         write!(f, "PaimonTableScan: table={}", self.table.identifier())?;
 
+        // These counts and the trace precede the filter/masking that run in
+        // `TableRead`, so EXPLAIN would disclose what enforcement hides — the
+        // same reason the statistics are suppressed.
+        if self.is_query_auth_restricted() {
+            write!(f, ", query-auth=restricted")?;
+            let columns = self
+                .read_type
+                .iter()
+                .map(|f| f.name().to_string())
+                .collect::<Vec<_>>();
+            return write!(f, ", projection=[{}]", columns.join(", "));
+        }
+
         let total_splits: usize = self.planned_partitions.iter().map(|p| p.len()).sum();
         let total_files: usize = self
             .planned_partitions

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

@@ -372,18 +372,24 @@ impl PaimonScanBuilder<'_> {
                 .collect()
         };
 
-        Ok(Arc::new(PaimonTableScan::new(
-            projected_schema,
-            self.table.clone(),
-            read_type,
-            self.pushed_predicate,
-            planned_partitions,
-            self.limit,
-            self.filter_exact,
-            self.scan_trace,
-            None,
-            self.case_sensitive,
-        )))
+        // From the plan, not its splits: a fully pruned plan carries no split
+        // to stamp, but its scan metadata is just as pre-enforcement.
+        let restricted = self.plan.planned_under_restricted_grant();
+        Ok(Arc::new(
+            PaimonTableScan::new(
+                projected_schema,
+                self.table.clone(),
+                read_type,
+                self.pushed_predicate,
+                planned_partitions,
+                self.limit,
+                self.filter_exact,
+                self.scan_trace,
+                None,
+                self.case_sensitive,
+            )
+            .with_query_auth_restricted(restricted),
+        ))
     }
 }
 
@@ -450,7 +456,11 @@ impl TableProvider for PaimonTableProvider {
             .map_err(to_datafusion_error)?;
 
         let target = state.config_options().execution.target_partitions;
+        // Inexact plan row counts (a query-auth row filter drops rows inside
+        // `TableRead`) would let DataFusion's aggregate-statistics rule answer
+        // COUNT(*) with the unfiltered count without ever invoking the read.
         let filter_exact = !filter_analysis.requires_residual
+            && plan.row_counts_exact()
             && filter_analysis
                 .pushed_predicate
                 .as_ref()

crates/integrations/datafusion/src/variant_pushdown.rs

@@ -243,23 +243,30 @@ impl ExtensionPlanner for VariantExtractionExtensionPlanner {
                 .collect()
         };
         let filter_exact = !filter_analysis.requires_residual
+            && plan.row_counts_exact()
             && filter_analysis
                 .pushed_predicate
                 .as_ref()
                 .is_none_or(|p| read_builder.is_exact_filter_pushdown(p));
 
-        Ok(Some(Arc::new(PaimonTableScan::new(
-            Arc::clone(&node.arrow_schema),
-            node.table.clone(),
-            node.read_type.clone(),
-            filter_analysis.pushed_predicate,
-            planned_partitions,
-            pushed_limit,
-            filter_exact,
-            Some(scan_trace),
-            Some(node.pushed_variants.clone()),
-            case_sensitive,
-        ))))
+        // From the plan, not its splits: a fully pruned plan carries no split to
+        // stamp, but its scan metadata is just as pre-enforcement.
+        let restricted = plan.planned_under_restricted_grant();
+        Ok(Some(Arc::new(
+            PaimonTableScan::new(
+                Arc::clone(&node.arrow_schema),
+                node.table.clone(),
+                node.read_type.clone(),
+                filter_analysis.pushed_predicate,
+                planned_partitions,
+                pushed_limit,
+                filter_exact,
+                Some(scan_trace),
+                Some(node.pushed_variants.clone()),
+                case_sensitive,
+            )
+            .with_query_auth_restricted(restricted),
+        )))
     }
 }