fix(auth): stop BLOB view predicate lookups at LIMIT under query-auth

A query-auth read with BLOB columns resolves the views its predicates
read before LIMIT applies, a whole batch at a time, so it could look up
references past the quota and fail on one no row needed. Read one
candidate at a time in that case, as the unrestricted read already does.

Author: plusplusjiajia

crates/paimon/src/table/table_read.rs

@@ -1169,17 +1169,29 @@ impl<'a> PaimonTableRead<'a> {
                     })
             });
         predicates.predicates = before_mask;
+        let blob_options = has_blobs.then(|| self.masked_blob_options(&masks));
         let unresolved_table;
-        if has_blobs {
+        if let Some(payload_options) = &blob_options {
             // Keep payload predicates above authorization too: they can open
             // a descriptor belonging to a row the server excludes.
             inner.data_predicates.clear();
-            unresolved_table = self
-                .table
-                .copy_with_options(std::collections::HashMap::from([
-                    ("blob-as-descriptor".to_string(), "true".to_string()),
-                    ("blob-view.resolve.enabled".to_string(), "false".to_string()),
-                ]));
+            let mut options = std::collections::HashMap::from([
+                ("blob-as-descriptor".to_string(), "true".to_string()),
+                ("blob-view.resolve.enabled".to_string(), "false".to_string()),
+            ]);
+            // Views a predicate reads resolve before LIMIT applies. Read one
+            // candidate at a time, as the unrestricted read does, so no
+            // reference past the quota is looked up.
+            let payload_options = CoreOptions::new(payload_options);
+            if self.limit.is_some()
+                && payload_options.blob_view_resolve_enabled()
+                && self.table.rest_env().is_some()
+                && !super::query_auth::leaf_names(&pending)
+                    .is_disjoint(&payload_options.blob_view_fields())
+            {
+                options.insert("read.batch-size".to_string(), "1".to_string());
+            }
+            unresolved_table = self.table.copy_with_options(options);
             inner.table = &unresolved_table;
         }
         let stream = inner.read_splits(data_splits, &inner.table.schema.core_options())?;
@@ -1191,7 +1203,6 @@ impl<'a> PaimonTableRead<'a> {
             row_filter_factory: None,
             file_fields: schema_fields.clone(),
         };
-        let blob_options = has_blobs.then(|| self.masked_blob_options(&masks));
         let stream = stream.map(move |batch| {
             let batch = batch?;
             let names_match = batch.num_columns() == physical.len()

crates/paimon/tests/rest_catalog_test.rs

@@ -1259,6 +1259,28 @@ async fn test_blob_view_limit_only_resolves_selected_references() {
         );
     }
 
+    // As without authorization, a predicate on the resolved view under LIMIT
+    // must stop before the invalid fourth reference the server still admits.
+    server.set_auth_response(
+        "default",
+        "blob_view_target",
+        rules(&[int_leaf(0, "id", "LESS_THAN", 5)], &[]),
+    );
+    let mut limited = authorized.new_read_builder();
+    limited.with_limit(2);
+    limited.with_filter(
+        PredicateBuilder::new(authorized.schema().fields())
+            .is_not_null("picture")
+            .unwrap(),
+    );
+    assert_eq!(
+        collect_blob_rows(&read_all(&limited).await.unwrap()),
+        vec![
+            (1, "Updated".to_string(), Some(first_payload.clone())),
+            (3, "Repeated again".to_string(), Some(b"bob".to_vec())),
+        ]
+    );
+
     // A view reference and its upstream BLOB can disagree on nullness.
     // Until rules can resolve that dependency safely, reject instead of
     // authorizing rows by the reference bytes.