This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 4f5e3abd fix(read): stop BLOB predicate reads at LIMIT (#903)
4f5e3abd is described below
commit 4f5e3abd41ab44c470b6c0ffbf17a29548b47676
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Sep 21 19:10:44 2026 +0800
fix(read): stop BLOB predicate reads at LIMIT (#903)
---
crates/paimon/src/table/data_evolution_reader.rs | 96 +++++++++++++++++++++++-
crates/paimon/tests/rest_catalog_test.rs | 45 ++++++++++-
2 files changed, 136 insertions(+), 5 deletions(-)
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 52de22c5..fb775ae0 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -248,6 +248,12 @@ impl DataEvolutionReader {
}
fn effective_batch_size(&self) -> Option<usize> {
+ if self.limit.is_some() && self.predicate_needs_blob_resolution() {
+ // A predicate on the resolved payload must inspect candidates in
+ // order. Reading more than one before the quota is updated can
+ // fetch a later, unselected payload (or invalid view reference).
+ return Some(1);
+ }
match (self.batch_size, self.limit) {
(Some(size), Some(limit)) if limit > 0 => Some(size.min(limit)),
(None, Some(limit)) if limit > 0 => Some(limit),
@@ -367,6 +373,26 @@ impl DataEvolutionReader {
Ok(selected)
}
+ fn predicate_needs_blob_resolution(&self) -> bool {
+ if self.predicates.is_empty() {
+ return false;
+ }
+ let mut resolved_fields = HashSet::new();
+ if self.blob_view_resolve_enabled && self.blob_view_rest_env.is_some()
{
+ resolved_fields.extend(self.blob_view_fields.iter().cloned());
+ }
+ if !self.blob_as_descriptor {
+
resolved_fields.extend(self.blob_descriptor_fields.iter().cloned());
+ resolved_fields.extend(
+ self.table_fields
+ .iter()
+ .filter(|field| field.data_type().is_blob_file_field())
+ .map(|field| field.name().to_string()),
+ );
+ }
+ predicates_reference_any_field(&self.predicates, &resolved_fields,
&self.table_fields)
+ }
+
pub(crate) fn with_parquet_read_budget(
mut self,
parquet_read_budget: Option<Arc<ReadBudget>>,
@@ -3033,6 +3059,54 @@ mod tests {
assert!(prefix_row_ranges(None, 10, 4, 0).is_empty());
}
+ #[test]
+ fn test_limit_uses_single_candidate_only_for_payload_predicates() {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let fields = vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(1, "payload".to_string(),
DataType::Blob(BlobType::new())),
+ ];
+ let predicate_builder = PredicateBuilder::new(&fields);
+ let reader = |predicate, blob_as_descriptor| {
+ DataEvolutionReader::new(
+ file_io.clone(),
+ SchemaManager::new(file_io.clone(),
"memory:/blob_batch_size".to_string()),
+ 1,
+ fields.clone(),
+ fields.clone(),
+ vec![predicate],
+ blob_as_descriptor,
+ HashSet::new(),
+ HashSet::new(),
+ false,
+ None,
+ )
+ .unwrap()
+ .with_batch_size(Some(8))
+ .with_limit(Some(3))
+ };
+
+ assert_eq!(
+ reader(predicate_builder.is_not_null("payload").unwrap(),
false).effective_batch_size(),
+ Some(1)
+ );
+ assert_eq!(
+ reader(predicate_builder.equal("id", Datum::Int(1)).unwrap(),
false)
+ .effective_batch_size(),
+ Some(3)
+ );
+ assert_eq!(
+ reader(predicate_builder.is_not_null("payload").unwrap(),
true).effective_batch_size(),
+ Some(3)
+ );
+ assert_eq!(
+ reader(predicate_builder.is_not_null("payload").unwrap(), false)
+ .with_limit(None)
+ .effective_batch_size(),
+ Some(8)
+ );
+ }
+
#[tokio::test]
async fn test_descriptor_columns_resolve_concurrently_and_preserve_order()
{
let schema = Arc::new(arrow_schema::Schema::new(vec![
@@ -5185,13 +5259,33 @@ mod tests {
let no_match = no_match_builder
.new_read()
.unwrap()
- .to_arrow(&[split])
+ .to_arrow(std::slice::from_ref(&split))
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
assert!(collect_int_values(&no_match, "id").is_empty());
+ // A predicate on the BLOB value must inspect each candidate, but
+ // the next batch may not fetch row 4 after three matches satisfy
+ // the quota. Its payload has a deliberately invalid checksum.
+ let mut blob_filter_builder = table.new_read_builder();
+ blob_filter_builder.with_limit(3);
+ blob_filter_builder.with_filter(
+ PredicateBuilder::new(table.schema().fields())
+ .is_not_null("payload")
+ .unwrap(),
+ );
+ let filtered = blob_filter_builder
+ .new_read()
+ .unwrap()
+ .to_arrow(&[split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(collect_int_values(&filtered, "id"), vec![1, 2, 3]);
+
// A BLOB-only raw-convertible split must obey the same quota.
builder.with_projection(&["payload"]).unwrap();
let raw_batches = builder
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index baa48cfe..d23309da 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -947,9 +947,14 @@ async fn
test_blob_view_limit_only_resolves_selected_references() {
write_batch(
&view,
blob_batch(
- vec![1, 2, 3],
- vec!["Kept", "Repeated", "Filtered"],
- vec![kept_ref.clone(), kept_ref, filtered_out_bad_ref],
+ vec![1, 2, 3, 4],
+ vec!["Kept", "Repeated", "Repeated again", "Filtered"],
+ vec![
+ kept_ref.clone(),
+ kept_ref.clone(),
+ kept_ref,
+ filtered_out_bad_ref,
+ ],
),
"view-writer",
)
@@ -998,7 +1003,7 @@ async fn
test_blob_view_limit_only_resolves_selected_references() {
vec![(1, "Kept".to_string(), Some(b"bob".to_vec()))]
);
- // LIMIT alone must not resolve the invalid reference in the third row.
+ // LIMIT alone must not resolve the invalid reference in the fourth row.
// The repeated reference is read from the lookup cache in a later batch.
let mut limited_builder = rest_view.new_read_builder();
limited_builder.with_limit(2);
@@ -1018,6 +1023,38 @@ async fn
test_blob_view_limit_only_resolves_selected_references() {
(2, "Repeated".to_string(), Some(b"bob".to_vec())),
]
);
+
+ // A predicate on the resolved view needs candidate lookups. With batches
+ // of two, the final batch must not resolve the invalid fourth reference
+ // after the third match reaches LIMIT.
+ let predicate_view = rest_view.copy_with_options(HashMap::from([(
+ "read.batch-size".to_string(),
+ "2".to_string(),
+ )]));
+ let mut filtered_builder = predicate_view.new_read_builder();
+ filtered_builder.with_limit(3);
+ filtered_builder.with_filter(
+ PredicateBuilder::new(predicate_view.schema().fields())
+ .is_not_null("picture")
+ .unwrap(),
+ );
+ let filtered_plan = filtered_builder.new_scan().plan().await.unwrap();
+ let filtered = filtered_builder
+ .new_read()
+ .unwrap()
+ .to_arrow(filtered_plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(
+ collect_blob_rows(&filtered),
+ vec![
+ (1, "Kept".to_string(), Some(b"bob".to_vec())),
+ (2, "Repeated".to_string(), Some(b"bob".to_vec())),
+ (3, "Repeated again".to_string(), Some(b"bob".to_vec())),
+ ]
+ );
}
#[cfg(not(windows))]
