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 b262b5f8 [core] Apply primary-key BLOB filters and limits before
payload resolution (#1053)
b262b5f8 is described below
commit b262b5f8d58ab28e018a14721a5c10a4aa71a951
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Oct 7 11:10:49 2026 +0800
[core] Apply primary-key BLOB filters and limits before payload resolution
(#1053)
---
.github/workflows/ci.yml | 9 +-
crates/paimon/src/table/data_evolution_reader.rs | 10 +-
crates/paimon/src/table/managed_blob_reader.rs | 694 ++++++++++++++++++---
.../paimon/src/table/managed_blob_table_tests.rs | 190 ++++++
crates/paimon/src/table/mod.rs | 1 +
crates/paimon/src/table/read_limit.rs | 28 +
crates/paimon/src/table/table_read.rs | 97 ++-
7 files changed, 940 insertions(+), 89 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 3138b024..d0cca364 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -118,11 +118,16 @@ jobs:
if: steps.release_license_inputs.outputs.changed == 'true'
run: python3 scripts/release_licenses.py --check
+ # Keep formatting and deny-warnings lints reproducible across stable
+ # releases. The build matrix independently checks the current toolchain.
+ - name: Install formatting and lint toolchain
+ run: rustup toolchain install 1.98.0 --profile minimal --component
rustfmt --component clippy
+
- name: Format
- run: cargo fmt --all -- --check
+ run: cargo +1.98.0 fmt --all -- --check
- name: Clippy
- run: cargo clippy --locked --all-targets --workspace --features
fulltext,vortex -- -D warnings
+ run: cargo +1.98.0 clippy --locked --all-targets --workspace
--features fulltext,vortex -- -D warnings
msrv:
name: msrv (${{ matrix.host }})
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 3e6ecb83..69c87940 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -22,6 +22,7 @@ use super::data_file_reader::{
append_null_row_id_column, attach_row_id, expand_selected_row_ids,
file_index_selection_to_local_ranges, insert_column_at,
DataFileReadTiming, DataFileReader,
};
+use super::read_limit::take_limited_batch;
use crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM;
use crate::arrow::format::FilePredicates;
use crate::arrow::format::MosaicPrefetchOptions;
@@ -141,15 +142,6 @@ pub(crate) struct DataEvolutionReader {
read_timing: Option<Arc<DataFileReadTiming>>,
}
-fn take_limited_batch(batch: RecordBatch, remaining: &mut Option<usize>) ->
RecordBatch {
- let Some(left) = remaining else {
- return batch;
- };
- let taken = batch.num_rows().min(*left);
- *left -= taken;
- batch.slice(0, taken)
-}
-
impl DataEvolutionReader {
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
diff --git a/crates/paimon/src/table/managed_blob_reader.rs
b/crates/paimon/src/table/managed_blob_reader.rs
index 0d311f95..c4e776b1 100644
--- a/crates/paimon/src/table/managed_blob_reader.rs
+++ b/crates/paimon/src/table/managed_blob_reader.rs
@@ -19,18 +19,20 @@
use super::blob_resolver::{resolve_descriptor_column, BlobReadLimiter};
use super::managed_blob_writer::{managed_blob_kind, ManagedBlobKind};
+use super::read_limit::take_limited_batch;
use super::{ArrowRecordBatchStream, Table};
use crate::arrow::format::FilePredicates;
use crate::io::FileIO;
-use crate::spec::{CoreOptions, DataField, Predicate};
+use crate::spec::{CoreOptions, DataField, Predicate, PredicateOperator};
use crate::Result;
-use arrow_array::builder::LargeBinaryBuilder;
use arrow_array::{
- Array, ArrayRef, LargeBinaryArray, ListArray, MapArray, RecordBatch,
StructArray,
+ Array, ArrayRef, BooleanArray, LargeBinaryArray, ListArray, MapArray,
RecordBatch, StructArray,
+ UInt32Array, UInt64Array,
};
+use arrow_buffer::{OffsetBuffer, ScalarBuffer};
use arrow_schema::DataType as ArrowDataType;
use futures::StreamExt;
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
use std::sync::Arc;
pub(crate) fn resolve_primary_key_blob_stream(
@@ -39,16 +41,19 @@ pub(crate) fn resolve_primary_key_blob_stream(
options: &CoreOptions<'_>,
file_io: FileIO,
parallelism: usize,
+ limit: Option<usize>,
) -> ArrowRecordBatchStream {
let selected = resolved_primary_key_blob_fields(fields, options);
- if selected.is_empty() {
+ if selected.is_empty() && limit.is_none() {
return stream;
}
let limiter = BlobReadLimiter::with_parallelism(parallelism);
Box::pin(async_stream::try_stream! {
let mut stream = stream;
- while let Some(batch) = stream.next().await {
- let batch = batch?;
+ let mut remaining = limit;
+ while remaining != Some(0) {
+ let Some(batch) = stream.next().await else { break };
+ let batch = take_limited_batch(batch?, &mut remaining);
yield resolve_batch(batch, &selected, &file_io, &limiter).await?;
}
})
@@ -81,10 +86,21 @@ fn resolved_blob_indices(fields: &[DataField], options:
&CoreOptions<'_>) -> Has
.collect()
}
-fn predicate_uses_resolved_blob(predicate: &Predicate, resolved:
&HashSet<usize>) -> bool {
- let mut referenced = HashSet::new();
- predicate.collect_leaf_field_indices(&mut referenced);
- referenced.iter().any(|index| resolved.contains(index))
+fn predicate_uses_resolved_blob(
+ predicate: &Predicate,
+ resolved: &HashSet<usize>,
+ fields: &[DataField],
+) -> bool {
+ let mut referenced = Vec::new();
+ crate::arrow::residual::collect_predicate_leaf_refs(predicate, &mut
referenced);
+ referenced.iter().any(|(name, index)| {
+ !crate::spec::is_row_tracking_column(name)
+ && fields
+ .iter()
+ .position(|field| field.name() == *name)
+ .or_else(|| fields.get(*index).map(|_| *index))
+ .is_some_and(|index| resolved.contains(&index))
+ })
}
/// Drop payload predicates from file and stats pruning while retaining safe
@@ -100,7 +116,9 @@ pub(crate) fn scan_predicates(table: &Table, predicates:
&[Predicate]) -> Vec<Pr
}
predicates
.iter()
- .filter(|predicate| !predicate_uses_resolved_blob(predicate,
&resolved))
+ .filter(|predicate| {
+ !predicate_uses_resolved_blob(predicate, &resolved,
table.schema().fields())
+ })
.cloned()
.collect()
}
@@ -126,7 +144,7 @@ impl ManagedBlobReadPlan {
}
predicates
.iter()
- .any(|predicate| predicate_uses_resolved_blob(predicate,
&resolved))
+ .any(|predicate| predicate_uses_resolved_blob(predicate,
&resolved, table_fields))
.then(|| {
let predicates = FilePredicates {
predicates: predicates.to_vec(),
@@ -153,35 +171,265 @@ impl ManagedBlobReadPlan {
options: &CoreOptions<'_>,
file_io: FileIO,
parallelism: usize,
+ limit: Option<usize>,
) -> ArrowRecordBatchStream {
- let stream = resolve_primary_key_blob_stream(
- stream,
- &self.scan_fields,
- options,
- file_io,
- parallelism,
- );
+ let resolved = resolved_primary_key_blob_fields(&self.scan_fields,
options);
+ let output_fields = resolved
+ .iter()
+ .copied()
+ .filter(|(index, _)| {
+ self.output_fields
+ .iter()
+ .any(|field| field.name() ==
self.scan_fields[*index].name())
+ })
+ .collect::<Vec<_>>();
+ let all_resolved = resolved_blob_indices(&self.predicates.file_fields,
options);
+ let ordinary_predicates = FilePredicates {
+ predicates: self
+ .predicates
+ .predicates
+ .iter()
+ .filter(|predicate| {
+ !predicate_uses_resolved_blob(
+ predicate,
+ &all_resolved,
+ &self.predicates.file_fields,
+ )
+ })
+ .cloned()
+ .collect(),
+ row_filter_factory: None,
+ file_fields: self.predicates.file_fields.clone(),
+ };
+ let limiter = BlobReadLimiter::with_parallelism(parallelism);
Box::pin(async_stream::try_stream! {
let mut stream = stream;
- while let Some(batch) = stream.next().await {
+ let mut remaining = limit;
+ 'batches: while remaining != Some(0) {
+ let Some(batch) = stream.next().await else { break };
let batch =
crate::arrow::residual::filter_record_batch_by_predicates(
- batch?, &self.predicates, &self.scan_fields,
+ batch?, &ordinary_predicates, &self.scan_fields,
)?;
- let indices = self.output_fields
- .iter()
- .map(|field| batch.schema().index_of(field.name()))
- .collect::<std::result::Result<Vec<_>, _>>()
- .map_err(|error| crate::Error::DataInvalid {
- message: format!("Managed BLOB output projection is
missing a column: {error}"),
+ // Inspect at most the remaining quota of candidate rows. Even
+ // if every candidate matches, none is beyond LIMIT. Rejected
+ // rows consume no quota; shrink the next batch accordingly.
+ let mut offset = 0;
+ while offset < batch.num_rows() {
+ if remaining == Some(0) { break 'batches; }
+ let width =
remaining.unwrap_or(batch.num_rows()).min(batch.num_rows() - offset);
+ let candidate = batch.slice(offset, width);
+ offset += width;
+ let mut filter = BlobPredicateBatch::new(
+ candidate, &self.predicates, &self.scan_fields,
+ &resolved, &file_io, &limiter,
+ );
+ let mut selected = vec![true; filter.batch.num_rows()];
+ for predicate in &self.predicates.predicates {
+ selected = filter.evaluate(predicate, selected).await?;
+ }
+ // Cache predicate payloads, and fetch output-only payloads
+ // solely for rows that survived Boolean short-circuiting.
+ filter.resolve_fields(&output_fields, &selected).await?;
+ let candidate = arrow_select::filter::filter_record_batch(
+ &filter.batch, &BooleanArray::from(selected),
+ ).map_err(|error| invalid(&error.to_string()))?;
+ let candidate = take_limited_batch(candidate, &mut
remaining);
+ if candidate.num_rows() == 0 { continue; }
+ let batch = candidate;
+ let indices = self.output_fields
+ .iter()
+ .map(|field| batch.schema().index_of(field.name()))
+ .collect::<std::result::Result<Vec<_>, _>>()
+ .map_err(|error| crate::Error::DataInvalid {
+ message: format!("Managed BLOB output projection
is missing a column: {error}"),
+ source: Some(Box::new(error)),
+ })?;
+ yield batch.project(&indices).map_err(|error|
crate::Error::DataInvalid {
+ message: format!("Failed to project managed BLOB read
output: {error}"),
source: Some(Box::new(error)),
})?;
- yield batch.project(&indices).map_err(|error|
crate::Error::DataInvalid {
- message: format!("Failed to project managed BLOB read
output: {error}"),
- source: Some(Box::new(error)),
- })?;
+ }
+ }
+ })
+ }
+}
+
+/// Java's descriptor rows resolve a payload only when a predicate accesses it.
+/// Apply the same AND/OR short-circuiting to batches, caching resolved rows so
+/// repeated leaves and output projection do not fetch a payload twice.
+struct BlobPredicateBatch<'a> {
+ batch: RecordBatch,
+ predicates: &'a FilePredicates,
+ scan_fields: &'a [DataField],
+ blob_fields: &'a [(usize, ManagedBlobKind)],
+ resolved: HashMap<usize, Vec<bool>>,
+ file_io: &'a FileIO,
+ limiter: &'a BlobReadLimiter,
+}
+
+impl<'a> BlobPredicateBatch<'a> {
+ fn new(
+ batch: RecordBatch,
+ predicates: &'a FilePredicates,
+ scan_fields: &'a [DataField],
+ blob_fields: &'a [(usize, ManagedBlobKind)],
+ file_io: &'a FileIO,
+ limiter: &'a BlobReadLimiter,
+ ) -> Self {
+ Self {
+ batch,
+ predicates,
+ scan_fields,
+ blob_fields,
+ resolved: HashMap::new(),
+ file_io,
+ limiter,
+ }
+ }
+
+ fn evaluate<'b>(
+ &'b mut self,
+ predicate: &'b Predicate,
+ active: Vec<bool>,
+ ) -> futures::future::BoxFuture<'b, Result<Vec<bool>>> {
+ Box::pin(async move {
+ if !active.iter().any(|selected| *selected) {
+ return Ok(active);
+ }
+ match predicate {
+ Predicate::AlwaysTrue => Ok(active),
+ Predicate::AlwaysFalse => Ok(vec![false; active.len()]),
+ Predicate::And(children) => {
+ let mut selected = active;
+ for child in children {
+ selected = self.evaluate(child, selected).await?;
+ }
+ Ok(selected)
+ }
+ Predicate::Or(children) => {
+ let mut pending = active;
+ let mut selected = vec![false; pending.len()];
+ for child in children {
+ let matched = self.evaluate(child,
pending.clone()).await?;
+ for row in 0..pending.len() {
+ selected[row] |= matched[row];
+ pending[row] &= !matched[row];
+ }
+ }
+ Ok(selected)
+ }
+ Predicate::Not(child) => {
+ let matched = self.evaluate(child, active.clone()).await?;
+ Ok(active
+ .iter()
+ .zip(matched)
+ .map(|(active, matched)| *active && !matched)
+ .collect())
+ }
+ Predicate::Leaf { op, .. } => {
+ let fields = self
+ .blob_fields
+ .iter()
+ .copied()
+ .filter(|(index, _)| {
+ predicate_uses_resolved_blob(
+ predicate,
+ &HashSet::from([*index]),
+ self.scan_fields,
+ )
+ })
+ .collect::<Vec<_>>();
+ // Java null checks inspect the BLOB reference's validity,
+ // without opening its payload. Resolution preserves nulls.
+ if !matches!(op, PredicateOperator::IsNull |
PredicateOperator::IsNotNull) {
+ self.resolve_fields(&fields, &active).await?;
+ }
+ let mask =
crate::arrow::residual::evaluate_predicates_mask(
+ &self.batch,
+ std::slice::from_ref(predicate),
+ &self.predicates.file_fields,
+ self.scan_fields,
+ )?;
+ Ok(active
+ .iter()
+ .enumerate()
+ .map(|(row, active)| {
+ *active && mask.as_ref().is_none_or(|mask|
mask.value(row))
+ })
+ .collect())
+ }
}
})
}
+
+ async fn resolve_fields(
+ &mut self,
+ fields: &[(usize, ManagedBlobKind)],
+ active: &[bool],
+ ) -> Result<()> {
+ for &(index, kind) in fields {
+ let resolved = self
+ .resolved
+ .entry(index)
+ .or_insert_with(|| vec![false; active.len()]);
+ let pending = active
+ .iter()
+ .zip(resolved.iter())
+ .map(|(active, resolved)| *active && !*resolved)
+ .collect::<Vec<_>>();
+ let count = pending.iter().filter(|pending| **pending).count();
+ if count == 0 {
+ continue;
+ }
+ if count == self.batch.num_rows() {
+ self.batch = resolve_batch(
+ self.batch.clone(),
+ &[(index, kind)],
+ self.file_io,
+ self.limiter,
+ )
+ .await?;
+ } else {
+ let selected = arrow_select::filter::filter_record_batch(
+ &self.batch,
+ &BooleanArray::from(pending.clone()),
+ )
+ .map_err(|error| invalid(&error.to_string()))?;
+ let selected =
+ resolve_batch(selected, &[(index, kind)], self.file_io,
self.limiter).await?;
+ let values = arrow_select::concat::concat(&[
+ self.batch.column(index).as_ref(),
+ selected.column(index).as_ref(),
+ ])
+ .map_err(|error| invalid(&error.to_string()))?;
+ let mut next = self.batch.num_rows() as u64;
+ let indices = UInt64Array::from(
+ pending
+ .iter()
+ .enumerate()
+ .map(|(row, selected)| {
+ if *selected {
+ let index = next;
+ next += 1;
+ index
+ } else {
+ row as u64
+ }
+ })
+ .collect::<Vec<_>>(),
+ );
+ let mut columns = self.batch.columns().to_vec();
+ columns[index] = arrow_select::take::take(values.as_ref(),
&indices, None)
+ .map_err(|error| invalid(&error.to_string()))?;
+ self.batch = RecordBatch::try_new(self.batch.schema(), columns)
+ .map_err(|error| invalid(&error.to_string()))?;
+ }
+ for (resolved, pending) in resolved.iter_mut().zip(pending) {
+ *resolved |= pending;
+ }
+ }
+ Ok(())
+ }
}
async fn resolve_batch(
@@ -190,6 +438,9 @@ async fn resolve_batch(
file_io: &FileIO,
limiter: &BlobReadLimiter,
) -> Result<RecordBatch> {
+ if fields.is_empty() || batch.num_rows() == 0 {
+ return Ok(batch);
+ }
let mut columns = batch.columns().to_vec();
for &(index, kind) in fields {
let column = match kind {
@@ -204,16 +455,17 @@ async fn resolve_batch(
.downcast_ref::<ListArray>()
.ok_or_else(|| invalid("ARRAY<BLOB> requires ListArray"))?;
let values = blob_values(array.values().as_ref())?;
- let visible = visible_child_values(values,
array.value_offsets(), array)?;
- let resolved =
- Arc::new(resolve_descriptor_column(&visible, file_io,
limiter.clone()).await?);
+ let visible = visible_child_values(values, array.offsets(),
array)?;
+ let resolved = Arc::new(
+ resolve_descriptor_column(&visible.values, file_io,
limiter.clone()).await?,
+ );
let ArrowDataType::List(element) = array.data_type() else {
unreachable!()
};
Arc::new(
ListArray::try_new(
element.clone(),
- array.offsets().clone(),
+ visible.offsets,
resolved,
array.nulls().cloned(),
)
@@ -226,25 +478,30 @@ async fn resolve_batch(
.downcast_ref::<MapArray>()
.ok_or_else(|| invalid("MAP<X, BLOB> requires MapArray"))?;
let values = blob_values(map.entries().column(1).as_ref())?;
- let visible = visible_child_values(values,
map.value_offsets(), map)?;
- let resolved =
- Arc::new(resolve_descriptor_column(&visible, file_io,
limiter.clone()).await?);
+ let visible = visible_child_values(values, map.offsets(),
map)?;
+ let resolved = Arc::new(
+ resolve_descriptor_column(&visible.values, file_io,
limiter.clone()).await?,
+ );
let ArrowDataType::Map(entries_field, ordered) =
map.data_type() else {
unreachable!()
};
let ArrowDataType::Struct(entry_fields) =
entries_field.data_type() else {
unreachable!()
};
- let entries = StructArray::try_new(
- entry_fields.clone(),
- vec![map.entries().column(0).clone(), resolved],
- None,
- )
- .map_err(|error| invalid(&error.to_string()))?;
+ let keys = match visible.indices {
+ Some(indices) => {
+
arrow_select::take::take(map.entries().column(0).as_ref(), &indices, None)
+ .map_err(|error| invalid(&error.to_string()))?
+ }
+ None => map.entries().column(0).clone(),
+ };
+ let entries =
+ StructArray::try_new(entry_fields.clone(), vec![keys,
resolved], None)
+ .map_err(|error| invalid(&error.to_string()))?;
Arc::new(
MapArray::try_new(
entries_field.clone(),
- map.offsets().clone(),
+ visible.offsets,
entries,
map.nulls().cloned(),
*ordered,
@@ -258,40 +515,59 @@ async fn resolve_batch(
RecordBatch::try_new(batch.schema(), columns).map_err(|error|
invalid(&error.to_string()))
}
-/// Arrow may retain child values beneath a null collection parent. They are
-/// invisible to the row and must not trigger descriptor I/O or fail the read
-/// when a stale URI happens to remain in an unused child slot.
+struct VisibleBlobChildren {
+ values: LargeBinaryArray,
+ offsets: OffsetBuffer<i32>,
+ indices: Option<UInt32Array>,
+}
+
+/// Drop children outside sliced parents or beneath NULL parents. Rebase the
+/// offsets and select the same map keys. Filling hidden values with NULL would
+/// violate a non-nullable element/value type, despite those slots being
unused.
fn visible_child_values(
values: &LargeBinaryArray,
- offsets: &[i32],
+ offsets: &OffsetBuffer<i32>,
parents: &dyn Array,
-) -> Result<LargeBinaryArray> {
+) -> Result<VisibleBlobChildren> {
if offsets.len() != parents.len() + 1 {
return Err(invalid("BLOB collection offsets do not match parent
rows"));
}
- let mut visible = vec![false; values.len()];
- for row in 0..parents.len() {
- if !parents.is_valid(row) {
- continue;
- }
- let start = usize::try_from(offsets[row])
- .map_err(|_| invalid("Negative BLOB collection offset"))?;
- let end = usize::try_from(offsets[row + 1])
- .map_err(|_| invalid("Negative BLOB collection offset"))?;
- let range = visible
- .get_mut(start..end)
- .ok_or_else(|| invalid("BLOB collection offset exceeds child
values"))?;
- range.fill(true);
+ if parents.null_count() == 0
+ && offsets[0] == 0
+ && offsets[parents.len()] as usize == values.len()
+ {
+ return Ok(VisibleBlobChildren {
+ values: values.clone(),
+ offsets: offsets.clone(),
+ indices: None,
+ });
}
- let mut builder = LargeBinaryBuilder::new();
- for (index, visible) in visible.into_iter().enumerate() {
- if visible && values.is_valid(index) {
- builder.append_value(values.value(index));
- } else {
- builder.append_null();
+ let mut indices = Vec::new();
+ let mut output_offsets = vec![0_i32];
+ for row in 0..parents.len() {
+ if parents.is_valid(row) {
+ let start = usize::try_from(offsets[row])
+ .map_err(|_| invalid("Negative BLOB collection offset"))?;
+ let end = usize::try_from(offsets[row + 1])
+ .map_err(|_| invalid("Negative BLOB collection offset"))?;
+ if start > end || end > values.len() {
+ return Err(invalid("BLOB collection offset exceeds child
values"));
+ }
+ indices.extend((start..end).map(|index| index as u32));
}
+ output_offsets.push(
+ i32::try_from(indices.len())
+ .map_err(|_| invalid("BLOB collection exceeds i32 offsets"))?,
+ );
}
- Ok(builder.finish())
+ let indices = UInt32Array::from(indices);
+ let selected = arrow_select::take::take(values, &indices, None)
+ .map_err(|error| invalid(&error.to_string()))?;
+ Ok(VisibleBlobChildren {
+ values: blob_values(selected.as_ref())?.clone(),
+ offsets: OffsetBuffer::new(ScalarBuffer::from(output_offsets)),
+ indices: Some(indices),
+ })
}
fn blob_values(array: &dyn Array) -> Result<&LargeBinaryArray> {
@@ -317,6 +593,278 @@ mod tests {
use arrow_buffer::{NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow_schema::{Field, Schema};
+ #[tokio::test]
+ async fn payload_filter_short_circuits_boolean_branches_and_null_checks() {
+ use crate::spec::{BlobType, DataType, Datum, IntType,
PredicateBuilder};
+ use futures::TryStreamExt;
+
+ let io = FileIOBuilder::new("memory").build().unwrap();
+ io.new_output("memory:/predicate-selected")
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"yes"))
+ .await
+ .unwrap();
+ let good = BlobDescriptor::new("memory:/predicate-selected".into(), 0,
3).serialize();
+ let missing = BlobDescriptor::new("memory:/must-not-open".into(), 0,
3).serialize();
+ let fields = vec![
+ DataField::new(0, "id".into(), DataType::Int(IntType::new())),
+ DataField::new(1, "payload".into(),
DataType::Blob(BlobType::new())),
+ ];
+ let batch = RecordBatch::try_new(
+ crate::arrow::build_target_arrow_schema(&fields).unwrap(),
+ vec![
+ Arc::new(arrow_array::Int32Array::from(vec![1, 2, 3])),
+ Arc::new(LargeBinaryArray::from(vec![
+ Some(missing.as_slice()),
+ Some(good.as_slice()),
+ None,
+ ])),
+ ],
+ )
+ .unwrap();
+ let builder = PredicateBuilder::new(&fields);
+ let equal_payload = builder
+ .equal("payload", Datum::Bytes(b"yes".to_vec()))
+ .unwrap();
+ let id_one = builder.equal("id", Datum::Int(1)).unwrap();
+ let cases = [
+ (
+ Predicate::Or(vec![id_one.clone(), equal_payload.clone()]),
+ vec![1, 2],
+ ),
+ (
+ Predicate::Or(vec![
+ Predicate::And(vec![
+ id_one.clone(),
+ builder.is_not_null("payload").unwrap(),
+ ]),
+ Predicate::And(vec![equal_payload.clone(),
equal_payload.clone()]),
+ ]),
+ vec![1, 2],
+ ),
+ (
+ Predicate::And(vec![
+ builder.is_not_null("payload").unwrap(),
+ Predicate::Not(Box::new(Predicate::Or(vec![
+ id_one,
+ builder
+ .not_equal("payload",
Datum::Bytes(b"yes".to_vec()))
+ .unwrap(),
+ ]))),
+ ]),
+ vec![2],
+ ),
+ (builder.is_null("payload").unwrap(), vec![3]),
+ (builder.is_not_null("payload").unwrap(), vec![1, 2]),
+ ];
+ let options = std::collections::HashMap::new();
+ let options = CoreOptions::new(&options);
+ for (predicate, expected) in cases {
+ for limit in [None, Some(1), Some(2)] {
+ let plan = ManagedBlobReadPlan::new(
+ &fields[..1],
+ std::slice::from_ref(&predicate),
+ &fields,
+ &options,
+ )
+ .unwrap();
+ let stream =
Box::pin(futures::stream::iter(vec![Ok(batch.clone())]));
+ let batches: Vec<RecordBatch> = plan
+ .finish(stream, &options, io.clone(), 2, limit)
+ .try_collect()
+ .await
+ .unwrap();
+ let actual: Vec<i32> = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<arrow_array::Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect();
+ assert_eq!(
+ actual,
+
expected[..limit.unwrap_or(expected.len()).min(expected.len())]
+ );
+ }
+ }
+ }
+
+ #[tokio::test]
+ async fn
predicate_and_output_share_payload_bytes_resembling_a_descriptor() {
+ use crate::spec::{BlobType, DataType, Datum, PredicateBuilder};
+ use futures::TryStreamExt;
+
+ let io = FileIOBuilder::new("memory").build().unwrap();
+ let payload =
+ BlobDescriptor::new("memory:/returned-bytes-must-not-open".into(),
0, 3).serialize();
+ io.new_output("memory:/predicate-payload")
+ .unwrap()
+ .write(bytes::Bytes::from(payload.clone()))
+ .await
+ .unwrap();
+ let descriptor =
+ BlobDescriptor::new("memory:/predicate-payload".into(), 0,
payload.len() as i64)
+ .serialize();
+ let fields = vec![DataField::new(
+ 0,
+ "payload".into(),
+ DataType::Blob(BlobType::new()),
+ )];
+ let batch = RecordBatch::try_new(
+ crate::arrow::build_target_arrow_schema(&fields).unwrap(),
+ vec![Arc::new(LargeBinaryArray::from(vec![Some(
+ descriptor.as_slice(),
+ )]))],
+ )
+ .unwrap();
+ let predicate = PredicateBuilder::new(&fields)
+ .equal("payload", Datum::Bytes(payload.clone()))
+ .unwrap();
+ let options = std::collections::HashMap::new();
+ let options = CoreOptions::new(&options);
+ let plan = ManagedBlobReadPlan::new(
+ &fields,
+ &[Predicate::And(vec![predicate.clone(), predicate])],
+ &fields,
+ &options,
+ )
+ .unwrap();
+ let stream = Box::pin(futures::stream::iter(vec![Ok(batch)]));
+ let batches: Vec<RecordBatch> = plan
+ .finish(stream, &options, io, 2, None)
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+ assert_eq!(
+ batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap()
+ .value(0),
+ payload
+ );
+ }
+
+ #[tokio::test]
+ async fn
collection_limit_skips_unselected_children_with_both_nullabilities() {
+ use crate::spec::{ArrayType, BlobType, DataType, MapType, VarCharType};
+ use futures::TryStreamExt;
+ for nullable in [true, false] {
+ let io = FileIOBuilder::new("memory").build().unwrap();
+ io.new_output("memory:/selected")
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"selected"))
+ .await
+ .unwrap();
+ let selected = BlobDescriptor::new("memory:/selected".into(), 0,
8).serialize();
+ let missing = BlobDescriptor::new("memory:/missing".into(), 0,
8).serialize();
+ let fields = vec![
+ DataField::new(
+ 0,
+ "items".into(),
+
DataType::Array(ArrayType::new(DataType::Blob(BlobType::with_nullable(
+ nullable,
+ )))),
+ ),
+ DataField::new(
+ 1,
+ "named".into(),
+ DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::string_type()),
+ DataType::Blob(BlobType::with_nullable(nullable)),
+ )),
+ ),
+ ];
+ let schema =
crate::arrow::build_target_arrow_schema(&fields).unwrap();
+ let values: ArrayRef = Arc::new(LargeBinaryArray::from(vec![
+ Some(selected.as_slice()),
+ Some(missing.as_slice()),
+ ]));
+ let ArrowDataType::List(element) = schema.field(0).data_type()
else {
+ panic!("expected list")
+ };
+ let offsets = OffsetBuffer::new(ScalarBuffer::from(vec![0, 1, 2]));
+ let list: ArrayRef = Arc::new(
+ ListArray::try_new(element.clone(), offsets.clone(),
values.clone(), None).unwrap(),
+ );
+ let ArrowDataType::Map(entries_field, ordered) =
schema.field(1).data_type() else {
+ panic!("expected map")
+ };
+ let ArrowDataType::Struct(entry_fields) =
entries_field.data_type() else {
+ panic!("expected entries")
+ };
+ let entries = StructArray::try_new(
+ entry_fields.clone(),
+ vec![Arc::new(StringArray::from(vec!["keep", "skip"])),
values],
+ None,
+ )
+ .unwrap();
+ let map: ArrayRef = Arc::new(
+ MapArray::try_new(entries_field.clone(), offsets, entries,
None, *ordered).unwrap(),
+ );
+ let batch = RecordBatch::try_new(schema, vec![list, map]).unwrap();
+ let stream = Box::pin(futures::stream::iter(vec![Ok(batch)]));
+ let batches: Vec<_> = resolve_primary_key_blob_stream(
+ stream,
+ &fields,
+ &CoreOptions::new(&std::collections::HashMap::new()),
+ io,
+ 2,
+ Some(1),
+ )
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+ assert_eq!(batches[0].num_rows(), 1);
+ let list = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .unwrap();
+ let items = list.value(0);
+ assert_eq!(
+ items
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap()
+ .value(0),
+ b"selected"
+ );
+ let map = batches[0]
+ .column(1)
+ .as_any()
+ .downcast_ref::<MapArray>()
+ .unwrap();
+ let entries = map.value(0);
+ assert_eq!(
+ entries
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .value(0),
+ "keep"
+ );
+ assert_eq!(
+ entries
+ .column(1)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap()
+ .value(0),
+ b"selected"
+ );
+ }
+ }
+
#[tokio::test]
async fn null_array_parent_does_not_fetch_hidden_descriptor() {
let missing =
@@ -353,7 +901,7 @@ mod tests {
.downcast_ref::<ListArray>()
.unwrap();
assert!(array.is_null(0));
- assert!(array.values().is_null(0));
+ assert_eq!(array.values().len(), 0);
}
#[tokio::test]
@@ -409,6 +957,6 @@ mod tests {
.downcast_ref::<MapArray>()
.unwrap();
assert!(map.is_null(0));
- assert!(map.entries().column(1).is_null(0));
+ assert_eq!(map.entries().len(), 0);
}
}
diff --git a/crates/paimon/src/table/managed_blob_table_tests.rs
b/crates/paimon/src/table/managed_blob_table_tests.rs
index e9d7cd92..85de4a08 100644
--- a/crates/paimon/src/table/managed_blob_table_tests.rs
+++ b/crates/paimon/src/table/managed_blob_table_tests.rs
@@ -1150,6 +1150,7 @@ async fn read_pending_values(table: &Table, file:
&crate::spec::DataFileMeta) ->
&table.schema().core_options(),
table.file_io().clone(),
2,
+ None,
)
.try_collect()
.await
@@ -1321,6 +1322,195 @@ async fn
primary_key_blob_payload_limit_keeps_later_buckets() {
}
}
+#[tokio::test]
+async fn primary_key_blob_limit_never_fetches_later_descriptors() {
+ for payload_filter in [false, true] {
+ for batch_size in [1, 8192] {
+ let io = test_file_io();
+ let path =
format!("memory:/pk_blob_limit_{payload_filter}_{batch_size}");
+ setup_dirs(&io, &path).await;
+ let table = scalar_table(
+ &io,
+ &path,
+ &[
+ ("blob-descriptor-field", "payload"),
+ ("read.batch-size", &batch_size.to_string()),
+ ],
+ );
+ io.new_output("memory:/limit-source")
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"selected"))
+ .await
+ .unwrap();
+ let selected = BlobDescriptor::new("memory:/limit-source".into(),
0, 8).serialize();
+ let missing = BlobDescriptor::new("memory:/not-selected".into(),
0, 8).serialize();
+ let mut writer = TableWrite::new(&table,
"limit-test".into()).unwrap();
+ writer
+ .write_arrow_batch(&scalar_batch(&[(1, Some(&selected)), (2,
Some(&missing))]))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "limit-test".into())
+ .commit(messages)
+ .await
+ .unwrap();
+ let mut builder = table.new_read_builder();
+ builder
+ .with_projection(&["id", "payload"])
+ .unwrap()
+ .with_limit(1);
+ if payload_filter {
+ builder.with_filter(
+ PredicateBuilder::new(table.schema().fields())
+ .equal("payload", Datum::Bytes(b"selected".to_vec()))
+ .unwrap(),
+ );
+ }
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+
assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ let row = batches.iter().find(|batch| batch.num_rows() !=
0).unwrap();
+ assert_eq!(
+ row.column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 1
+ );
+ assert_eq!(
+ row.column(1)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap()
+ .value(0),
+ b"selected"
+ );
+ let streaming: Vec<_> = plan
+ .splits()
+ .iter()
+ .map(|split| {
+ super::DataSplitBuilder::new()
+ .with_snapshot(split.snapshot_id())
+ .with_partition(split.partition().clone())
+ .with_bucket(split.bucket())
+ .with_bucket_path(split.bucket_path().to_string())
+ .with_total_buckets(split.total_buckets())
+ .with_data_files(split.data_files().to_vec())
+ .with_streaming(true)
+ .build()
+ .unwrap()
+ })
+ .collect();
+ for mixed in [false, true] {
+ let mut splits = streaming.clone();
+ if mixed {
+ splits.extend_from_slice(plan.splits());
+ }
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow_with_row_kind(&splits)
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+
assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ let row = batches.iter().find(|batch| batch.num_rows() !=
0).unwrap();
+ assert_eq!(
+ row.column_by_name("payload")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap()
+ .value(0),
+ b"selected"
+ );
+ assert_eq!(
+ row.column_by_name(crate::spec::ROW_KIND_FIELD_NAME)
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .value(0),
+ "+I"
+ );
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn primary_key_blob_sequence_predicate_does_not_alias_payload_index() {
+ let io = test_file_io();
+ let path = "memory:/pk_blob_sequence_filter";
+ setup_dirs(&io, path).await;
+ let table = scalar_table(&io, path, &[("blob-descriptor-field",
"payload")]);
+ io.new_output("memory:/sequence-selected")
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"selected"))
+ .await
+ .unwrap();
+ let selected = BlobDescriptor::new("memory:/sequence-selected".into(), 0,
8).serialize();
+ let missing = BlobDescriptor::new("memory:/sequence-unselected".into(), 0,
8).serialize();
+ let mut writer = TableWrite::new(&table, "sequence-test".into()).unwrap();
+ writer
+ .write_arrow_batch(&scalar_batch(&[(1, Some(&selected)), (2,
Some(&missing))]))
+ .await
+ .unwrap();
+ TableCommit::new(table.clone(), "sequence-test".into())
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ // Metadata indices belong to the caller's read schema. Index 1 here
+ // coincides with payload in the logical table schema; the name is
decisive.
+ let predicate_fields = [
+ table.schema().fields()[0].clone(),
+ DataField::new(
+ SEQUENCE_NUMBER_FIELD_ID,
+ SEQUENCE_NUMBER_FIELD_NAME.into(),
+ DataType::BigInt(BigIntType::new()),
+ ),
+ ];
+ let predicate = PredicateBuilder::new(&predicate_fields)
+ .equal(SEQUENCE_NUMBER_FIELD_NAME, Datum::Long(0))
+ .unwrap();
+ let mut builder = table.new_read_builder();
+ builder
+ .with_projection(&["payload"])
+ .unwrap()
+ .with_filter(predicate);
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ assert_eq!(
+ batches
+ .iter()
+ .find(|batch| batch.num_rows() != 0)
+ .unwrap()
+ .column(0)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap()
+ .value(0),
+ b"selected"
+ );
+}
+
#[tokio::test]
async fn postpone_blob_failed_checkpoint_preserves_prepared_packs() {
let file_io = test_file_io();
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 58134550..09240586 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -119,6 +119,7 @@ mod postpone_retract;
mod prepared_files;
mod query_auth;
mod read_builder;
+mod read_limit;
pub mod referenced_files;
pub(crate) mod rest_env;
pub(crate) mod row_id_predicate;
diff --git a/crates/paimon/src/table/read_limit.rs
b/crates/paimon/src/table/read_limit.rs
new file mode 100644
index 00000000..39823009
--- /dev/null
+++ b/crates/paimon/src/table/read_limit.rs
@@ -0,0 +1,28 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use arrow_array::RecordBatch;
+
+/// Consume the quota after merging and exact filtering. Slicing before output
+/// BLOB resolution matches Java's lazy BlobDescriptorResolvingRow and
+/// LimitRecordReader: discarded payloads remain untouched.
+pub(super) fn take_limited_batch(batch: RecordBatch, remaining: &mut
Option<usize>) -> RecordBatch {
+ let Some(left) = remaining else { return batch };
+ let taken = batch.num_rows().min(*left);
+ *left -= taken;
+ batch.slice(0, taken)
+}
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index ee0d3a0f..0d45ea63 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -172,9 +172,8 @@ impl<'a> TableRead<'a> {
})
}
- /// Pass the read limit to paths that can enforce it. Data-evolution reads
- /// apply it before BLOB resolution; other Paimon reads still use the
- /// builder limit only as a scan hint.
+ /// Pass the limit to Data Evolution and primary-key readers. These paths
+ /// apply it after merging/filtering and before output BLOB resolution.
pub(crate) fn with_limit(self, limit: Option<usize>) -> Self {
match self.0 {
TableReadKind::Paimon(mut read) => {
@@ -514,6 +513,9 @@ impl<'a> PaimonTableRead<'a> {
.iter()
.cloned()
.partition(DataSplit::is_streaming);
+ if self.limit.is_some() && !streaming.is_empty() &&
!materialized.is_empty() {
+ return self.limited_mixed_row_kind_stream(materialized, streaming);
+ }
let mut streams = Vec::with_capacity(2);
if !materialized.is_empty() {
streams.push(prepend_insert_row_kind_stream(
@@ -534,6 +536,47 @@ impl<'a> PaimonTableRead<'a> {
Ok(Box::pin(stream::select_all(streams)))
}
+ /// Open each kind of split only after the preceding group consumed its
+ /// quota. Truncating two independently limited streams afterward could
+ /// fetch BLOB payloads from a group whose rows will never be returned.
+ fn limited_mixed_row_kind_stream(
+ &self,
+ materialized: Vec<DataSplit>,
+ streaming: Vec<DataSplit>,
+ ) -> crate::Result<ArrowRecordBatchStream> {
+ let table = self.table.clone();
+ let read_type = self.read_type.clone();
+ let data_predicates = self.data_predicates.clone();
+ let row_filter_factory = self.row_filter_factory.clone();
+ let parquet_read_budget = self.parquet_read_budget()?;
+ let resources = self.resources.clone();
+ let data_file_read_timing = self.data_file_read_timing.clone();
+ let blob_parallelism = self.blob_parallelism;
+ let limit = self.limit;
+ Ok(Box::pin(async_stream::try_stream! {
+ let mut remaining = limit;
+ for group in [materialized, streaming] {
+ if remaining == Some(0) { break; }
+ let read = PaimonTableRead {
+ table: &table,
+ read_type: read_type.clone(),
+ data_predicates: data_predicates.clone(),
+ row_filter_factory: row_filter_factory.clone(),
+ parquet_read_budget: Some(parquet_read_budget.clone()),
+ resources: resources.clone(),
+ data_file_read_timing: data_file_read_timing.clone(),
+ blob_parallelism,
+ limit: remaining,
+ };
+ let mut stream = read.to_arrow_with_row_kind(&group)?;
+ while remaining != Some(0) {
+ let Some(batch) = stream.next().await else { break };
+ yield super::read_limit::take_limited_batch(batch?, &mut
remaining);
+ }
+ }
+ }))
+ }
+
fn audit_raw_stream(
&self,
plan: &IncrementalPlan,
@@ -554,6 +597,8 @@ impl<'a> PaimonTableRead<'a> {
has_value_kind: bool,
include_sequence: bool,
) -> crate::Result<ArrowRecordBatchStream> {
+ use super::managed_blob_reader::{resolve_primary_key_blob_stream,
ManagedBlobReadPlan};
+
let core_options = self.table.schema().core_options();
let user_read_type = self.read_type.clone();
let audit_schema = audit_schema_for_read_type(&user_read_type,
include_sequence)?;
@@ -577,14 +622,34 @@ impl<'a> PaimonTableRead<'a> {
));
}
+ let primary_key = !self.table.schema().primary_keys().is_empty();
+ let blob_plan = primary_key
+ .then(|| {
+ ManagedBlobReadPlan::new(
+ &read_type,
+ &self.data_predicates,
+ self.table.schema().fields(),
+ &core_options,
+ )
+ })
+ .flatten();
+ let scan_type = blob_plan
+ .as_ref()
+ .map_or_else(|| read_type.clone(), |plan|
plan.scan_fields().to_vec());
+ let predicates = if blob_plan.is_some() {
+ Vec::new()
+ } else {
+ self.data_predicates.clone()
+ };
let reader = DataFileReader::new(
self.table.file_io.clone(),
self.table.schema_manager().clone(),
self.table.schema().id(),
self.table.schema.fields().to_vec(),
- read_type,
- self.data_predicates.clone(),
+ scan_type,
+ predicates,
)
+ .with_blob_as_descriptor(primary_key)
.with_file_index_read_enabled(core_options.file_index_read_enabled())
.with_batch_size(Some(core_options.read_batch_size()?))
.with_blob_parallelism(self.blob_parallelism)
@@ -592,6 +657,26 @@ impl<'a> PaimonTableRead<'a> {
.with_table_options(self.table.schema().options().clone())
.with_mosaic_prefetch(configured_mosaic_prefetch(self.table)?);
let raw_stream = reader.read(data_splits)?;
+ let raw_stream = if let Some(plan) = blob_plan {
+ plan.finish(
+ raw_stream,
+ &core_options,
+ self.table.file_io.clone(),
+ self.blob_parallelism,
+ self.limit,
+ )
+ } else if primary_key {
+ resolve_primary_key_blob_stream(
+ raw_stream,
+ &read_type,
+ &core_options,
+ self.table.file_io.clone(),
+ self.blob_parallelism,
+ self.limit,
+ )
+ } else {
+ raw_stream
+ };
Ok(Box::pin(async_stream::try_stream! {
futures::pin_mut!(raw_stream);
@@ -1003,6 +1088,7 @@ impl<'a> PaimonTableRead<'a> {
core_options,
self.table.file_io.clone(),
self.blob_parallelism,
+ self.limit,
));
}
@@ -1013,6 +1099,7 @@ impl<'a> PaimonTableRead<'a> {
core_options,
self.table.file_io.clone(),
self.blob_parallelism,
+ self.limit,
))
}
