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 539cf430 feat(python): configure native BLOB read parallelism (#871)
539cf430 is described below
commit 539cf430806feddbea7fb6b60494fae4921009b3
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Sep 18 16:27:06 2026 +0800
feat(python): configure native BLOB read parallelism (#871)
---
.../python/python/pypaimon_rust/datafusion.pyi | 3 +
bindings/python/src/read.rs | 24 ++++++
bindings/python/tests/test_read.py | 13 ++++
bindings/python/tests/test_write.py | 4 +-
crates/paimon/src/arrow/format/blob.rs | 90 ++++++++++++++++++++--
crates/paimon/src/arrow/format/mod.rs | 17 ++--
crates/paimon/src/table/blob_resolver.rs | 54 ++++++++++++-
crates/paimon/src/table/data_evolution_reader.rs | 33 ++++++--
.../table/data_evolution_reader/blob_fallback.rs | 16 +++-
crates/paimon/src/table/data_file_reader.rs | 13 ++++
crates/paimon/src/table/format_read_builder.rs | 13 +++-
crates/paimon/src/table/format_table_read.rs | 10 +++
crates/paimon/src/table/read_builder.rs | 34 +++++++-
crates/paimon/src/table/table_read.rs | 45 ++++++++++-
14 files changed, 335 insertions(+), 34 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index e9aadcf6..8bf6b3cd 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -84,6 +84,9 @@ class ReadBuilder:
"""Set a scan-planning row-limit hint, not an exact cap: a matching
split is
returned whole. Apply application-level limiting for an exact bound."""
...
+ def with_blob_parallelism(self, blob_parallelism: int) -> "ReadBuilder":
+ """Set the maximum number of concurrent BLOB range reads. Must be
positive."""
+ ...
def with_filter(self, predicate: dict) -> "ReadBuilder": ...
def with_row_ranges(self, ranges: Sequence[tuple[int, int]]) ->
"ReadBuilder":
"""Set Data Evolution row ranges. Empty selects no rows; format tables
are unsupported."""
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index 0a5ab881..42c89165 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -112,6 +112,7 @@ pub struct PyReadBuilder {
filter: Option<Predicate>,
row_ranges: Option<Vec<RowRange>>,
case_sensitive: bool,
+ blob_parallelism: Option<usize>,
}
impl PyReadBuilder {
@@ -123,6 +124,7 @@ impl PyReadBuilder {
filter: None,
row_ranges: None,
case_sensitive: true,
+ blob_parallelism: None,
}
}
@@ -168,6 +170,7 @@ impl PyReadBuilder {
filter: None,
row_ranges: None,
case_sensitive: true,
+ blob_parallelism: None,
})
}
}
@@ -184,6 +187,20 @@ impl PyReadBuilder {
slf
}
+ /// Set the maximum number of concurrent BLOB range reads for this read.
+ fn with_blob_parallelism(
+ mut slf: PyRefMut<'_, Self>,
+ blob_parallelism: usize,
+ ) -> PyResult<PyRefMut<'_, Self>> {
+ if blob_parallelism == 0 {
+ return Err(PyValueError::new_err(
+ "blob_parallelism must be greater than zero",
+ ));
+ }
+ slf.blob_parallelism = Some(blob_parallelism);
+ Ok(slf)
+ }
+
/// Set whether column-name matching (projection and predicate resolution)
is
/// case-sensitive. Defaults to `true` (exact match); pass `false` to match
/// column names case-insensitively (ASCII case-folding).
@@ -256,6 +273,7 @@ impl PyReadBuilder {
limit: self.limit,
filter: self.filter.clone(),
case_sensitive: self.case_sensitive,
+ blob_parallelism: self.blob_parallelism,
}
}
}
@@ -378,6 +396,7 @@ pub struct PyTableRead {
limit: Option<usize>,
filter: Option<Predicate>,
case_sensitive: bool,
+ blob_parallelism: Option<usize>,
}
impl PyTableRead {
@@ -396,6 +415,11 @@ impl PyTableRead {
&self.filter,
self.case_sensitive,
)?;
+ if let Some(blob_parallelism) = self.blob_parallelism {
+ builder
+ .with_blob_parallelism(blob_parallelism)
+ .map_err(to_py_err)?;
+ }
// Validate config (e.g. projection) before the empty-splits fast
// path so an invalid projection fails consistently regardless of
// how many splits are passed.
diff --git a/bindings/python/tests/test_read.py
b/bindings/python/tests/test_read.py
index 448cc47a..e772f73d 100644
--- a/bindings/python/tests/test_read.py
+++ b/bindings/python/tests/test_read.py
@@ -66,6 +66,19 @@ def test_with_limit():
assert plan is not None
+def test_with_blob_parallelism():
+ with tempfile.TemporaryDirectory() as warehouse:
+ table = _make_table_with_data(warehouse)
+ builder = table.new_read_builder().with_blob_parallelism(2)
+ plan = builder.new_scan().plan()
+ assert pa.Table.from_batches(
+ builder.new_read().read(plan.splits())
+ ).num_rows == 3
+
+ with pytest.raises(ValueError, match="must be greater than zero"):
+ table.new_read_builder().with_blob_parallelism(0)
+
+
def test_with_row_ranges():
with tempfile.TemporaryDirectory() as warehouse:
ctx = SQLContext()
diff --git a/bindings/python/tests/test_write.py
b/bindings/python/tests/test_write.py
index ed5be8b3..98d9274d 100644
--- a/bindings/python/tests/test_write.py
+++ b/bindings/python/tests/test_write.py
@@ -299,7 +299,9 @@ def test_ltz_schema_alias_write_roundtrip(tmp_path,
precision, unit, micros, fra
schema["fields"][1]["type"] = "TIMESTAMP_LTZ({})".format(precision)
schema_path.write_text(json.dumps(schema))
table = _get_table(str(tmp_path))
- assert "LocalZonedTimestamp" in table.schema().fields()[1].field_type()
+ assert table.schema().fields()[1].field_type() == (
+ f"TIMESTAMP({precision}) WITH LOCAL TIME ZONE"
+ )
value = datetime(2026, 9, 15, 20, 0, 0, micros, tzinfo=timezone.utc)
arrow_schema = pa.schema([("id", pa.int32()), ("ts", pa.timestamp(unit,
tz="UTC"))])
builder = table.new_write_builder()
diff --git a/crates/paimon/src/arrow/format/blob.rs
b/crates/paimon/src/arrow/format/blob.rs
index 6002d465..f03db782 100644
--- a/crates/paimon/src/arrow/format/blob.rs
+++ b/crates/paimon/src/arrow/format/blob.rs
@@ -39,6 +39,7 @@ use std::sync::Arc;
pub(crate) struct BlobFormatReader {
descriptor_mode: bool,
file_path: String,
+ blob_parallelism: usize,
}
impl BlobFormatReader {
@@ -46,8 +47,15 @@ impl BlobFormatReader {
Self {
descriptor_mode,
file_path,
+ blob_parallelism: DEFAULT_BLOB_READ_PARALLELISM,
}
}
+
+ pub(crate) fn with_blob_parallelism(mut self, blob_parallelism: usize) ->
Self {
+ debug_assert!(blob_parallelism > 0);
+ self.blob_parallelism = blob_parallelism;
+ self
+ }
}
pub(crate) struct IndexedBlobReader {
@@ -55,21 +63,42 @@ pub(crate) struct IndexedBlobReader {
index: BlobFileIndex,
descriptor_mode: bool,
file_path: String,
+ blob_parallelism: usize,
}
impl IndexedBlobReader {
+ #[cfg(test)]
pub(crate) async fn open(
reader: Box<dyn FileRead>,
file_size: u64,
file_path: String,
descriptor_mode: bool,
) -> crate::Result<Self> {
+ Self::open_with_parallelism(
+ reader,
+ file_size,
+ file_path,
+ descriptor_mode,
+ DEFAULT_BLOB_READ_PARALLELISM,
+ )
+ .await
+ }
+
+ pub(crate) async fn open_with_parallelism(
+ reader: Box<dyn FileRead>,
+ file_size: u64,
+ file_path: String,
+ descriptor_mode: bool,
+ blob_parallelism: usize,
+ ) -> crate::Result<Self> {
+ debug_assert!(blob_parallelism > 0);
let index = BlobFileIndex::load(reader.as_ref(), file_size).await?;
Ok(Self {
reader,
index,
descriptor_mode,
file_path,
+ blob_parallelism,
})
}
@@ -85,7 +114,7 @@ impl IndexedBlobReader {
build_descriptor_values(&self.index, positions, &self.file_path)
} else {
let planned_reads = plan_blob_reads(&self.index, positions)?;
- fetch_blob_values(self.reader.as_ref(), planned_reads).await
+ fetch_blob_values(self.reader.as_ref(), planned_reads,
self.blob_parallelism).await
}
}
@@ -99,6 +128,7 @@ impl IndexedBlobReader {
planned_reads,
&self.file_path,
self.descriptor_mode,
+ self.blob_parallelism,
)
.await
}
@@ -115,6 +145,7 @@ impl IndexedBlobReader {
&self.file_path,
self.descriptor_mode,
key_type,
+ self.blob_parallelism,
)
.await
}
@@ -137,7 +168,7 @@ const BLOB_INLINE_HEADER_SIZE: u64 = 4;
const BLOB_TRAILER_SIZE: u64 = 12;
const BLOB_ENTRY_OVERHEAD: u64 = BLOB_INLINE_HEADER_SIZE + BLOB_TRAILER_SIZE;
const DEFAULT_BATCH_SIZE: usize = 128;
-const BLOB_READ_CONCURRENCY: usize = 8;
+pub(crate) const DEFAULT_BLOB_READ_PARALLELISM: usize = 8;
const BLOB_ARRAY_MAGIC_NUMBER: i32 = 1094861634;
const BLOB_ARRAY_VERSION: u8 = 1;
const BLOB_ARRAY_HEADER_SIZE: u64 = 9;
@@ -178,11 +209,12 @@ impl FormatFileReader for BlobFormatReader {
let target_schema = build_target_arrow_schema(read_fields)?;
let batch_size = batch_size.unwrap_or(DEFAULT_BATCH_SIZE);
- let blob_reader = IndexedBlobReader::open(
+ let blob_reader = IndexedBlobReader::open_with_parallelism(
reader,
file_size,
self.file_path.clone(),
self.descriptor_mode,
+ self.blob_parallelism,
)
.await?;
let mut selection = RowSelectionCursor::new(blob_reader.num_rows(),
row_selection)?;
@@ -653,6 +685,7 @@ fn plan_blob_reads(
async fn fetch_blob_values(
reader: &dyn FileRead,
planned_reads: Vec<PlannedBlobRead>,
+ blob_parallelism: usize,
) -> crate::Result<Vec<BlobReadValue>> {
futures::stream::iter(planned_reads.into_iter().map(|planned_read| async
move {
match planned_read {
@@ -663,7 +696,7 @@ async fn fetch_blob_values(
.map(BlobReadValue::Value),
}
}))
- .buffered(BLOB_READ_CONCURRENCY)
+ .buffered(blob_parallelism)
.try_collect()
.await
}
@@ -759,6 +792,7 @@ async fn fetch_blob_array_values(
planned_reads: Vec<PlannedBlobArrayRead>,
file_path: &str,
descriptor_mode: bool,
+ blob_parallelism: usize,
) -> crate::Result<Vec<BlobReadValue>> {
futures::stream::iter(planned_reads.into_iter().map(|planned_read| async
move {
match planned_read {
@@ -774,7 +808,7 @@ async fn fetch_blob_array_values(
}
}
}))
- .buffered(BLOB_READ_CONCURRENCY)
+ .buffered(blob_parallelism)
.try_collect()
.await
}
@@ -1076,6 +1110,7 @@ async fn fetch_blob_map_values(
file_path: &str,
descriptor_mode: bool,
key_type: &DataType,
+ blob_parallelism: usize,
) -> crate::Result<Vec<BlobReadValue>> {
futures::stream::iter(planned_reads.into_iter().map(|planned_read| async
move {
match planned_read {
@@ -1087,7 +1122,7 @@ async fn fetch_blob_map_values(
}
}
}))
- .buffered(BLOB_READ_CONCURRENCY)
+ .buffered(blob_parallelism)
.try_collect()
.await
}
@@ -2446,7 +2481,48 @@ mod tests {
.collect::<Vec<_>>()
);
assert!(reader.max_in_flight() > 1);
- assert!(reader.max_in_flight() <= BLOB_READ_CONCURRENCY);
+ assert!(reader.max_in_flight() <= DEFAULT_BLOB_READ_PARALLELISM);
+ }
+
+ #[tokio::test]
+ async fn test_blob_reader_honors_configured_parallelism() {
+ let payloads = (0_u8..12).map(|value| vec![value]).collect::<Vec<_>>();
+ let rows = payloads
+ .iter()
+ .map(|payload| Some(payload.as_slice()))
+ .collect::<Vec<_>>();
+ let file_bytes = blob_test_utils::build_blob_file_bytes(&rows);
+ let read_fields = vec![DataField::new(
+ 0,
+ "payload".to_string(),
+ DataType::Blob(BlobType::new()),
+ )];
+ let reader = TrackingFileRead::new(Bytes::from(file_bytes.clone()));
+
+ let batches = BlobFormatReader::new(String::new(), false)
+ .with_blob_parallelism(2)
+ .read_batch_stream(
+ Box::new(reader.clone()),
+ file_bytes.len() as u64,
+ &read_fields,
+ None,
+ Some(12),
+ None,
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(batches.len(), 1);
+ assert_eq!(
+ collect_binary_values(&batches[0]),
+ (0_u8..12)
+ .map(|value| Some(vec![value]))
+ .collect::<Vec<_>>()
+ );
+ assert_eq!(reader.max_in_flight(), 2);
}
#[tokio::test]
diff --git a/crates/paimon/src/arrow/format/mod.rs
b/crates/paimon/src/arrow/format/mod.rs
index f67ae565..17d21c9f 100644
--- a/crates/paimon/src/arrow/format/mod.rs
+++ b/crates/paimon/src/arrow/format/mod.rs
@@ -190,7 +190,13 @@ pub(crate) fn create_format_reader(
blob_as_descriptor: bool,
read_fields: &[DataField],
) -> crate::Result<Box<dyn FormatFileReader>> {
- create_format_reader_with_budget(path, blob_as_descriptor, read_fields,
None)
+ create_format_reader_with_budget(
+ path,
+ blob_as_descriptor,
+ read_fields,
+ None,
+ blob::DEFAULT_BLOB_READ_PARALLELISM,
+ )
}
/// Create a format reader with a scan-shared Parquet resource budget.
@@ -199,6 +205,7 @@ pub(crate) fn create_format_reader_with_budget(
blob_as_descriptor: bool,
read_fields: &[DataField],
parquet_read_budget: Option<Arc<ParquetReadBudget>>,
+ blob_parallelism: usize,
) -> crate::Result<Box<dyn FormatFileReader>> {
let lower = path.to_ascii_lowercase();
let reader: Box<dyn FormatFileReader> = if lower.ends_with(".parquet") {
@@ -207,10 +214,10 @@ pub(crate) fn create_format_reader_with_budget(
None => parquet::ParquetFormatReader::default(),
})
} else if lower.ends_with(".blob") {
- Box::new(blob::BlobFormatReader::new(
- path.to_string(),
- blob_as_descriptor,
- ))
+ Box::new(
+ blob::BlobFormatReader::new(path.to_string(), blob_as_descriptor)
+ .with_blob_parallelism(blob_parallelism),
+ )
} else if lower.ends_with(".orc") {
Box::new(orc::OrcFormatReader)
} else if lower.ends_with(".avro") {
diff --git a/crates/paimon/src/table/blob_resolver.rs
b/crates/paimon/src/table/blob_resolver.rs
index 22cee1b0..e66e4dcc 100644
--- a/crates/paimon/src/table/blob_resolver.rs
+++ b/crates/paimon/src/table/blob_resolver.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM;
use crate::io::{FileIO, FileRead};
use crate::spec::BlobDescriptor;
use crate::Result;
@@ -29,7 +30,7 @@ use tokio::sync::{OwnedSemaphorePermit, Semaphore};
const BLOB_RANGE_MERGE_GAP: u64 = 64 * 1024;
const BLOB_RANGE_MERGE_MAX_SPAN: u64 = 8 * 1024 * 1024;
-pub(crate) const BLOB_DESCRIPTOR_READ_CONCURRENCY: usize = 8;
+pub(crate) const BLOB_DESCRIPTOR_READ_CONCURRENCY: usize =
DEFAULT_BLOB_READ_PARALLELISM;
const BLOB_DESCRIPTOR_READ_BYTE_UNIT: u64 = 1024 * 1024;
const BLOB_DESCRIPTOR_READ_MAX_IN_FLIGHT_BYTES: u64 = 64 * 1024 * 1024;
@@ -105,6 +106,7 @@ impl BlobReader {
}
let limiter = self.limiter.clone();
+ let blob_parallelism = limiter.parallelism();
let groups: Vec<Vec<(usize, Vec<u8>)>> = stream::iter(by_uri)
.map(|(uri, entries)| {
let limiter = limiter.clone();
@@ -145,7 +147,7 @@ impl BlobReader {
)
}
})
- .buffer_unordered(BLOB_DESCRIPTOR_READ_CONCURRENCY)
+ .buffer_unordered(blob_parallelism)
.try_collect()
.await?;
@@ -322,6 +324,7 @@ pub(crate) struct BlobReadLimiter {
bytes: Arc<Semaphore>,
byte_unit: u64,
max_byte_permits: u32,
+ parallelism: usize,
}
impl BlobReadLimiter {
@@ -333,6 +336,18 @@ impl BlobReadLimiter {
)
}
+ pub(crate) fn with_parallelism(parallelism: usize) -> Self {
+ Self::with_limits(
+ parallelism,
+ BLOB_DESCRIPTOR_READ_MAX_IN_FLIGHT_BYTES,
+ BLOB_DESCRIPTOR_READ_BYTE_UNIT,
+ )
+ }
+
+ pub(crate) fn parallelism(&self) -> usize {
+ self.parallelism
+ }
+
fn with_limits(request_limit: usize, byte_budget: u64, byte_unit: u64) ->
Self {
assert!(request_limit > 0);
assert!(byte_budget > 0);
@@ -344,6 +359,7 @@ impl BlobReadLimiter {
bytes: Arc::new(Semaphore::new(max_byte_permits as usize)),
byte_unit,
max_byte_permits: max_byte_permits as u32,
+ parallelism: request_limit,
}
}
@@ -574,6 +590,7 @@ async fn read_blob_groups(
groups: Vec<BlobReadGroup>,
limiter: BlobReadLimiter,
) -> Result<Vec<ResolvedMergedBlobRead>> {
+ let blob_parallelism = limiter.parallelism();
let grouped_results: Vec<Vec<ResolvedMergedBlobRead>> =
stream::iter(groups)
.map(|group| {
@@ -582,7 +599,7 @@ async fn read_blob_groups(
read_merged_blob_ranges(&group.uri, group.reader,
group.reads, limiter).await
}
})
- .buffer_unordered(BLOB_DESCRIPTOR_READ_CONCURRENCY)
+ .buffer_unordered(blob_parallelism)
.try_collect()
.await?;
Ok(grouped_results.into_iter().flatten().collect())
@@ -594,6 +611,7 @@ async fn read_merged_blob_ranges(
reads: Vec<MergedBlobRead>,
limiter: BlobReadLimiter,
) -> Result<Vec<ResolvedMergedBlobRead>> {
+ let blob_parallelism = limiter.parallelism();
stream::iter(reads)
.map(|merged| {
let uri = uri.to_string();
@@ -627,7 +645,7 @@ async fn read_merged_blob_ranges(
Ok(ResolvedMergedBlobRead { merged, data })
}
})
- .buffer_unordered(BLOB_DESCRIPTOR_READ_CONCURRENCY)
+ .buffer_unordered(blob_parallelism)
.try_collect()
.await
}
@@ -768,6 +786,34 @@ mod tests {
assert!(reader.max_in_flight() <= BLOB_DESCRIPTOR_READ_CONCURRENCY);
}
+ #[tokio::test]
+ async fn test_blob_range_reads_honor_configured_parallelism() {
+ let reader =
TrackingFileRead::new(Bytes::from_static(b"abcdefghijkl"));
+ let reads = (0..12)
+ .map(|row| MergedBlobRead {
+ start: row,
+ end: row + 1,
+ requests: vec![BlobReadRequest {
+ row: row as usize,
+ offset: row,
+ length: 1,
+ }],
+ })
+ .collect();
+
+ let results = read_merged_blob_ranges(
+ "memory:/blob.bin",
+ std::sync::Arc::new(reader.clone()),
+ reads,
+ BlobReadLimiter::with_parallelism(2),
+ )
+ .await
+ .unwrap();
+
+ assert_eq!(results.len(), 12);
+ assert_eq!(reader.max_in_flight(), 2);
+ }
+
#[tokio::test]
async fn test_blob_range_reads_apply_byte_budget_and_preserve_rows() {
let reader = TrackingFileRead::new(Bytes::from_static(b"abcdefgh"));
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index a456505a..a4480878 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -17,11 +17,12 @@
mod blob_fallback;
-use super::blob_resolver::{BlobReadLimiter, BLOB_DESCRIPTOR_READ_CONCURRENCY};
+use super::blob_resolver::BlobReadLimiter;
use super::data_file_reader::{
append_null_row_id_column, attach_row_id, expand_selected_row_ids,
insert_column_at,
DataFileReadTiming, DataFileReader,
};
+use crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM;
use crate::arrow::format::FilePredicates;
use crate::arrow::{build_target_arrow_schema, ParquetReadBudget};
use crate::deletion_vector::{DeletionVector, DeletionVectorFactory};
@@ -112,6 +113,7 @@ pub(crate) struct DataEvolutionReader {
blob_view_resolve_enabled: bool,
blob_view_rest_env: Option<RESTEnv>,
blob_read_limiter: BlobReadLimiter,
+ blob_parallelism: usize,
batch_size: Option<usize>,
parquet_read_budget: Option<Arc<ParquetReadBudget>>,
read_timing: Option<Arc<DataFileReadTiming>>,
@@ -190,6 +192,7 @@ impl DataEvolutionReader {
blob_view_resolve_enabled,
blob_view_rest_env,
blob_read_limiter: BlobReadLimiter::new(),
+ blob_parallelism: DEFAULT_BLOB_READ_PARALLELISM,
batch_size: None,
parquet_read_budget: None,
read_timing: None,
@@ -201,6 +204,13 @@ impl DataEvolutionReader {
self
}
+ pub(crate) fn with_blob_parallelism(mut self, blob_parallelism: usize) ->
Self {
+ debug_assert!(blob_parallelism > 0);
+ self.blob_read_limiter =
BlobReadLimiter::with_parallelism(blob_parallelism);
+ self.blob_parallelism = blob_parallelism;
+ self
+ }
+
pub(crate) fn with_parquet_read_budget(
mut self,
parquet_read_budget: Option<Arc<ParquetReadBudget>>,
@@ -255,6 +265,7 @@ impl DataEvolutionReader {
},
)
.with_batch_size(self.batch_size)
+ .with_blob_parallelism(self.blob_parallelism)
.with_parquet_read_budget(self.parquet_read_budget.clone())
.with_read_timing(self.read_timing.clone());
@@ -617,6 +628,7 @@ impl DataEvolutionReader {
let table_fields = self.table_fields.clone();
let blob_descriptor_fields = self.blob_descriptor_fields.clone();
let blob_as_descriptor = self.blob_as_descriptor;
+ let blob_parallelism = self.blob_parallelism;
let batch_size = self.batch_size;
let parquet_read_budget = self.parquet_read_budget.clone();
let read_timing = self.read_timing.clone();
@@ -705,6 +717,7 @@ impl DataEvolutionReader {
table_fields.clone(),
batch_size,
blob_as_descriptor,
+ blob_parallelism,
source_parquet_read_budget.clone(),
read_timing.clone(),
anchor_deletion_vector.as_ref(),
@@ -870,17 +883,23 @@ async fn resolve_descriptor_columns(
file_io: &FileIO,
limiter: &BlobReadLimiter,
) -> crate::Result<RecordBatch> {
- resolve_descriptor_columns_with(batch, blob_descriptor_fields, |column| {
+ resolve_descriptor_columns_with(
+ batch,
+ blob_descriptor_fields,
+ limiter.parallelism(),
+ |column| {
let file_io = file_io.clone();
let limiter = limiter.clone();
async move { super::blob_resolver::resolve_blob_column(&column,
&file_io, limiter).await }
- })
+ },
+ )
.await
}
async fn resolve_descriptor_columns_with<F, Fut>(
batch: RecordBatch,
blob_descriptor_fields: &HashSet<String>,
+ blob_parallelism: usize,
resolve: F,
) -> crate::Result<RecordBatch>
where
@@ -913,7 +932,7 @@ where
let future = resolve(column);
async move { future.await.map(|resolved| (idx, resolved)) }
})
- .buffer_unordered(BLOB_DESCRIPTOR_READ_CONCURRENCY)
+ .buffer_unordered(blob_parallelism)
.try_collect()
.await?;
for (idx, resolved) in resolved_columns {
@@ -1237,6 +1256,7 @@ fn open_source_stream(
table_fields: Vec<DataField>,
batch_size: Option<usize>,
blob_as_descriptor: bool,
+ blob_parallelism: usize,
parquet_read_budget: Option<Arc<ParquetReadBudget>>,
read_timing: Option<Arc<DataFileReadTiming>>,
anchor_deletion_vector: Option<&DeletionVectorContext>,
@@ -1272,6 +1292,7 @@ fn open_source_stream(
batch_size,
file_io,
blob_as_descriptor,
+ blob_parallelism,
anchor_deletion_vector.cloned(),
);
}
@@ -1287,6 +1308,7 @@ fn open_source_stream(
batch_size,
file_io,
blob_as_descriptor,
+ blob_parallelism,
anchor_deletion_vector.cloned(),
);
}
@@ -1303,6 +1325,7 @@ fn open_source_stream(
)
.with_batch_size(batch_size)
.with_blob_as_descriptor(blob_as_descriptor)
+ .with_blob_parallelism(blob_parallelism)
.with_parquet_read_budget(parquet_read_budget)
.with_read_timing(read_timing);
@@ -2686,7 +2709,7 @@ mod tests {
let in_flight = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let max_in_flight = Arc::new(std::sync::atomic::AtomicUsize::new(0));
- let resolved = resolve_descriptor_columns_with(batch, &fields,
|column| {
+ let resolved = resolve_descriptor_columns_with(batch, &fields, 2,
|column| {
let in_flight = in_flight.clone();
let max_in_flight = max_in_flight.clone();
async move {
diff --git a/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
b/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
index dbd6a0ae..c0674c2c 100644
--- a/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
+++ b/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
@@ -51,6 +51,7 @@ impl LazyBlobFile {
positions: &[usize],
file_io: &FileIO,
blob_as_descriptor: bool,
+ blob_parallelism: usize,
field_kind: &BlobFieldKind,
) -> crate::Result<Vec<BlobReadValue>> {
if self.reader.is_none() {
@@ -63,11 +64,12 @@ impl LazyBlobFile {
})?;
let input = file_io.new_input(&self.path)?;
let reader = input.reader().await?;
- let reader = IndexedBlobReader::open(
+ let reader = IndexedBlobReader::open_with_parallelism(
Box::new(reader),
file_size,
self.path.clone(),
blob_as_descriptor,
+ blob_parallelism,
)
.await?;
let indexed_rows =
@@ -116,6 +118,7 @@ pub(super) fn read(
batch_size: Option<usize>,
file_io: FileIO,
blob_as_descriptor: bool,
+ blob_parallelism: usize,
anchor_deletion_vector: Option<DeletionVectorContext>,
) -> crate::Result<ArrowRecordBatchStream> {
if read_fields.len() != 1 ||
!read_fields[0].data_type().is_blob_file_field() {
@@ -180,6 +183,7 @@ pub(super) fn read(
target_schema.clone(),
&file_io,
blob_as_descriptor,
+ blob_parallelism,
&field_kind,
).await?;
}
@@ -193,6 +197,7 @@ async fn resolve_batch(
target_schema: Arc<arrow_schema::Schema>,
file_io: &FileIO,
blob_as_descriptor: bool,
+ blob_parallelism: usize,
field_kind: &BlobFieldKind,
) -> crate::Result<RecordBatch> {
let mut resolved = (0..row_ids.len())
@@ -241,7 +246,13 @@ async fn resolve_batch(
if !file_positions.is_empty() {
let values = file
- .read_positions(&file_positions, file_io,
blob_as_descriptor, field_kind)
+ .read_positions(
+ &file_positions,
+ file_io,
+ blob_as_descriptor,
+ blob_parallelism,
+ field_kind,
+ )
.await?;
for (output_position, value) in
output_positions.into_iter().zip(values) {
if !matches!(&value, BlobReadValue::Placeholder) {
@@ -438,6 +449,7 @@ mod tests {
schema,
&file_io,
false,
+ crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM,
&BlobFieldKind::Scalar,
)
.await
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index e5d8e7de..b24f03ad 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -16,6 +16,7 @@
// under the License.
use crate::arrow::build_target_arrow_schema;
+use crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM;
use crate::arrow::format::create_format_reader_with_budget;
use crate::arrow::schema_evolution::{create_index_mapping, NULL_FIELD_INDEX};
use crate::arrow::ParquetReadBudget;
@@ -117,6 +118,7 @@ pub(crate) struct DataFileReader {
file_index_read_enabled: bool,
row_filter_factory: Option<Arc<dyn crate::arrow::RowFilterFactory>>,
blob_as_descriptor: bool,
+ blob_parallelism: usize,
batch_size: Option<usize>,
parquet_read_budget: Option<Arc<ParquetReadBudget>>,
read_timing: Option<Arc<DataFileReadTiming>>,
@@ -141,6 +143,7 @@ impl DataFileReader {
file_index_read_enabled: false,
row_filter_factory: None,
blob_as_descriptor: false,
+ blob_parallelism: DEFAULT_BLOB_READ_PARALLELISM,
batch_size: None,
parquet_read_budget: None,
read_timing: None,
@@ -152,6 +155,12 @@ impl DataFileReader {
self
}
+ pub(crate) fn with_blob_parallelism(mut self, blob_parallelism: usize) ->
Self {
+ debug_assert!(blob_parallelism > 0);
+ self.blob_parallelism = blob_parallelism;
+ self
+ }
+
pub(crate) fn with_file_index_read_enabled(mut self, enabled: bool) ->
Self {
self.file_index_read_enabled = enabled;
self
@@ -448,6 +457,7 @@ impl DataFileReader {
let file_io = self.file_io.clone();
let split = split.clone();
let blob_as_descriptor = self.blob_as_descriptor;
+ let blob_parallelism = self.blob_parallelism;
let batch_size = self.batch_size;
let parquet_read_budget = self.parquet_read_budget.clone();
let read_timing = self.read_timing.clone();
@@ -509,6 +519,7 @@ impl DataFileReader {
blob_as_descriptor,
&format_read_fields,
parquet_read_budget,
+ blob_parallelism,
)?;
let input_file = file_io.new_input(&path_to_read)?;
let open_start = read_timing.as_ref().map(|_| Instant::now());
@@ -716,6 +727,7 @@ impl DataFileReader {
let file_io = self.file_io.clone();
let split = split.clone();
let blob_as_descriptor = self.blob_as_descriptor;
+ let blob_parallelism = self.blob_parallelism;
let parquet_read_budget = self.parquet_read_budget.clone();
let target_schema = build_target_arrow_schema(&read_type)?;
@@ -781,6 +793,7 @@ impl DataFileReader {
blob_as_descriptor,
&format_read_fields,
parquet_read_budget,
+ blob_parallelism,
)?;
let input_file = file_io.new_input(&path_to_read)?;
let file_reader = input_file.reader().await?;
diff --git a/crates/paimon/src/table/format_read_builder.rs
b/crates/paimon/src/table/format_read_builder.rs
index a2b6f81f..f426e096 100644
--- a/crates/paimon/src/table/format_read_builder.rs
+++ b/crates/paimon/src/table/format_read_builder.rs
@@ -22,6 +22,7 @@ use super::read_builder::split_scan_predicates;
use super::read_builder::{resolve_projected_fields,
validate_projection_possible};
use super::table_read::configured_parquet_read_budget;
use super::{Table, TableRead, TableScan};
+use crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM;
use crate::arrow::ParquetReadBudget;
use crate::spec::{DataField, Predicate};
use crate::table::source::RowRange;
@@ -42,6 +43,7 @@ pub(crate) struct FormatReadBuilder<'a> {
row_ranges: Option<Vec<RowRange>>,
case_sensitive: bool,
parquet_read_budget: Option<Arc<ParquetReadBudget>>,
+ blob_parallelism: usize,
}
impl<'a> FormatReadBuilder<'a> {
@@ -56,6 +58,7 @@ impl<'a> FormatReadBuilder<'a> {
row_ranges: None,
case_sensitive: true,
parquet_read_budget: None,
+ blob_parallelism: DEFAULT_BLOB_READ_PARALLELISM,
}
}
@@ -115,6 +118,11 @@ impl<'a> FormatReadBuilder<'a> {
self
}
+ pub(crate) fn with_blob_parallelism(&mut self, blob_parallelism: usize) ->
&mut Self {
+ self.blob_parallelism = blob_parallelism;
+ self
+ }
+
pub(crate) fn new_scan(&self) -> TableScan<'a> {
TableScan::new(
self.table,
@@ -144,13 +152,14 @@ impl<'a> FormatReadBuilder<'a> {
Some(budget) => Arc::clone(budget),
None => configured_parquet_read_budget(self.table)?,
};
- Ok(TableRead::new_format(
+ TableRead::new_format(
self.table,
read_type,
self.data_predicates.clone(),
self.limit,
)
- .with_parquet_read_budget(parquet_read_budget))
+ .with_parquet_read_budget(parquet_read_budget)
+ .with_blob_parallelism(self.blob_parallelism)
}
/// Resolve the effective read type, deferring projection name resolution
to
diff --git a/crates/paimon/src/table/format_table_read.rs
b/crates/paimon/src/table/format_table_read.rs
index cb6e232f..c20c3839 100644
--- a/crates/paimon/src/table/format_table_read.rs
+++ b/crates/paimon/src/table/format_table_read.rs
@@ -21,6 +21,7 @@ use super::data_file_reader::DataFileReader;
use super::read_builder::split_scan_predicates;
use super::table_read::configured_parquet_read_budget;
use super::{ArrowRecordBatchStream, Table};
+use crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM;
use crate::arrow::partition::partition_array;
use crate::arrow::{build_target_arrow_schema, ParquetReadBudget};
use crate::spec::{DataField, Predicate};
@@ -38,6 +39,7 @@ pub(crate) struct FormatTableRead<'a> {
row_filter_factory: Option<Arc<dyn crate::arrow::RowFilterFactory>>,
parquet_read_budget: Option<Arc<ParquetReadBudget>>,
limit: Option<usize>,
+ blob_parallelism: usize,
}
impl<'a> FormatTableRead<'a> {
@@ -54,6 +56,7 @@ impl<'a> FormatTableRead<'a> {
row_filter_factory: None,
parquet_read_budget: None,
limit,
+ blob_parallelism: DEFAULT_BLOB_READ_PARALLELISM,
}
}
@@ -87,6 +90,11 @@ impl<'a> FormatTableRead<'a> {
self
}
+ pub(crate) fn with_blob_parallelism(mut self, blob_parallelism: usize) ->
Self {
+ self.blob_parallelism = blob_parallelism;
+ self
+ }
+
fn parquet_read_budget(&self) -> crate::Result<Arc<ParquetReadBudget>> {
match &self.parquet_read_budget {
Some(budget) => Ok(Arc::clone(budget)),
@@ -121,6 +129,7 @@ impl<'a> FormatTableRead<'a> {
let batch_size = Some(core_options.read_batch_size()?);
let row_filter_factory = self.row_filter_factory.clone();
let parquet_read_budget = Some(self.parquet_read_budget()?);
+ let blob_parallelism = self.blob_parallelism;
Ok(try_stream! {
for split in splits {
@@ -137,6 +146,7 @@ impl<'a> FormatTableRead<'a> {
data_predicates.clone(),
)
.with_batch_size(batch_size)
+ .with_blob_parallelism(blob_parallelism)
.with_parquet_read_budget(parquet_read_budget.clone());
if let Some(factory) = &row_filter_factory {
reader =
reader.with_row_filter_factory(Arc::clone(factory));
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index c258a1c5..f5087883 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -26,6 +26,7 @@ use super::incremental_scan::{IncrementalScan,
IncrementalScanMode};
use super::partition_filter::PartitionFilter;
use super::table_read::{configured_parquet_read_budget, TableRead};
use super::{Table, TableScan};
+use crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM;
use crate::spec::{CoreOptions, DataField, Predicate};
use crate::table::source::RowRange;
use crate::{Error, Result};
@@ -237,6 +238,25 @@ impl<'a> ReadBuilder<'a> {
self
}
+ /// Set the maximum number of concurrent BLOB range reads for the
resulting read.
+ pub fn with_blob_parallelism(&mut self, blob_parallelism: usize) ->
Result<&mut Self> {
+ if blob_parallelism == 0 {
+ return Err(Error::DataInvalid {
+ message: "BLOB read parallelism must be greater than
zero".to_string(),
+ source: None,
+ });
+ }
+ match &mut self.0 {
+ ReadBuilderKind::Paimon(builder) => {
+ builder.with_blob_parallelism(blob_parallelism);
+ }
+ ReadBuilderKind::Format(builder) => {
+ builder.with_blob_parallelism(blob_parallelism);
+ }
+ }
+ Ok(self)
+ }
+
/// Inject a Parquet budget shared with sibling scan partitions.
#[doc(hidden)]
pub fn with_parquet_read_budget(
@@ -317,6 +337,7 @@ struct PaimonReadBuilder<'a> {
derived_row_ranges: Option<Vec<RowRange>>,
case_sensitive: bool,
parquet_read_budget: Option<Arc<crate::arrow::ParquetReadBudget>>,
+ blob_parallelism: usize,
}
impl<'a> PaimonReadBuilder<'a> {
@@ -331,6 +352,7 @@ impl<'a> PaimonReadBuilder<'a> {
derived_row_ranges: None,
case_sensitive: true,
parquet_read_budget: None,
+ blob_parallelism: DEFAULT_BLOB_READ_PARALLELISM,
}
}
@@ -471,6 +493,11 @@ impl<'a> PaimonReadBuilder<'a> {
self
}
+ fn with_blob_parallelism(&mut self, blob_parallelism: usize) -> &mut Self {
+ self.blob_parallelism = blob_parallelism;
+ self
+ }
+
/// Create a table scan. Call [TableScan::plan] to get splits.
///
/// Projection names are resolved here on a best-effort basis: the resolved
@@ -519,10 +546,9 @@ impl<'a> PaimonReadBuilder<'a> {
Some(budget) => Arc::clone(budget),
None => configured_parquet_read_budget(self.table)?,
};
- Ok(
- TableRead::new(self.table, read_type,
self.filter.data_predicates.clone())
- .with_parquet_read_budget(parquet_read_budget),
- )
+ TableRead::new(self.table, read_type,
self.filter.data_predicates.clone())
+ .with_parquet_read_budget(parquet_read_budget)
+ .with_blob_parallelism(self.blob_parallelism)
}
/// Resolve the effective read type, deferring projection name resolution
to
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index b684d411..9d641841 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -23,6 +23,7 @@ use super::kv_file_reader::{KeyValueFileReader,
KeyValueReadConfig};
use super::read_builder::split_scan_predicates;
use super::{ArrowRecordBatchStream, Table};
use crate::arrow::build_target_arrow_schema;
+use crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM;
use crate::arrow::ParquetReadBudget;
use crate::spec::{
BigIntType, CoreOptions, DataField, DataType, MergeEngine, Predicate,
TinyIntType,
@@ -133,6 +134,24 @@ impl<'a> TableRead<'a> {
}
}
+ /// Set the maximum number of concurrent BLOB range reads for this read.
+ pub fn with_blob_parallelism(self, blob_parallelism: usize) ->
crate::Result<Self> {
+ if blob_parallelism == 0 {
+ return Err(crate::Error::DataInvalid {
+ message: "BLOB read parallelism must be greater than
zero".to_string(),
+ source: None,
+ });
+ }
+ Ok(match self.0 {
+ TableReadKind::Paimon(read) => Self(TableReadKind::Paimon(
+ read.with_blob_parallelism(blob_parallelism),
+ )),
+ TableReadKind::Format(read) => Self(TableReadKind::Format(
+ read.with_blob_parallelism(blob_parallelism),
+ )),
+ })
+ }
+
/// Attach an engine-specific Parquet decoder-filter factory.
///
/// The hook is used only by schema-identical raw reads. Callers must still
@@ -230,6 +249,7 @@ struct PaimonTableRead<'a> {
row_filter_factory: Option<Arc<dyn crate::arrow::RowFilterFactory>>,
parquet_read_budget: Option<Arc<ParquetReadBudget>>,
data_file_read_timing: Option<Arc<DataFileReadTiming>>,
+ blob_parallelism: usize,
}
impl<'a> PaimonTableRead<'a> {
@@ -246,6 +266,7 @@ impl<'a> PaimonTableRead<'a> {
row_filter_factory: None,
parquet_read_budget: None,
data_file_read_timing: None,
+ blob_parallelism: DEFAULT_BLOB_READ_PARALLELISM,
}
}
@@ -294,6 +315,11 @@ impl<'a> PaimonTableRead<'a> {
self
}
+ fn with_blob_parallelism(mut self, blob_parallelism: usize) -> Self {
+ self.blob_parallelism = blob_parallelism;
+ self
+ }
+
fn parquet_read_budget(&self) -> crate::Result<Arc<ParquetReadBudget>> {
match &self.parquet_read_budget {
Some(budget) => Ok(Arc::clone(budget)),
@@ -337,6 +363,7 @@ impl<'a> PaimonTableRead<'a> {
let read_type = self.read_type.clone();
let data_predicates = self.data_predicates.clone();
let parquet_read_budget = self.parquet_read_budget()?;
+ let blob_parallelism = self.blob_parallelism;
Ok(Box::pin(async_stream::try_stream! {
let mut workers = stream::iter(pairs.into_iter().map(|(before,
after)| {
@@ -346,7 +373,8 @@ impl<'a> PaimonTableRead<'a> {
let parquet_read_budget = Arc::clone(&parquet_read_budget);
let worker: ArrowRecordBatchStream =
Box::pin(async_stream::try_stream! {
let pair_read = PaimonTableRead::new(&table, read_type,
data_predicates)
- .with_parquet_read_budget(parquet_read_budget);
+ .with_parquet_read_budget(parquet_read_budget)
+ .with_blob_parallelism(blob_parallelism);
let mut pair_stream =
pair_read.to_diff_after_image_stream(&before, &after)?;
while let Some(batch) = pair_stream.next().await {
yield batch?;
@@ -421,6 +449,7 @@ impl<'a> PaimonTableRead<'a> {
)
.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)
.with_parquet_read_budget(Some(self.parquet_read_budget()?));
let raw_stream = reader.read(&data_splits)?;
@@ -480,6 +509,7 @@ impl<'a> PaimonTableRead<'a> {
let read_type = self.read_type.clone();
let data_predicates = self.data_predicates.clone();
let parquet_read_budget = self.parquet_read_budget()?;
+ let blob_parallelism = self.blob_parallelism;
Ok(Box::pin(async_stream::try_stream! {
let mut workers = stream::iter(pairs.into_iter().map(|(before,
after)| {
@@ -489,7 +519,8 @@ impl<'a> PaimonTableRead<'a> {
let parquet_read_budget = Arc::clone(&parquet_read_budget);
let worker: ArrowRecordBatchStream =
Box::pin(async_stream::try_stream! {
let pair_read = PaimonTableRead::new(&table, read_type,
data_predicates)
- .with_parquet_read_budget(parquet_read_budget);
+ .with_parquet_read_budget(parquet_read_budget)
+ .with_blob_parallelism(blob_parallelism);
let mut pair_stream =
pair_read.to_audit_log_arrow_for_diff(&before,
&after)?;
while let Some(batch) = pair_stream.next().await {
@@ -535,11 +566,13 @@ impl<'a> PaimonTableRead<'a> {
let read_type_for_output = self.read_type.clone();
let data_predicates = self.data_predicates.clone();
let parquet_read_budget = self.parquet_read_budget()?;
+ let blob_parallelism = self.blob_parallelism;
Ok(Box::pin(async_stream::try_stream! {
let core_options = CoreOptions::new(table.schema().options());
let pair_read = PaimonTableRead::new(&table,
diff_read_type.clone(), data_predicates)
- .with_parquet_read_budget(parquet_read_budget);
+ .with_parquet_read_budget(parquet_read_budget)
+ .with_blob_parallelism(blob_parallelism);
let before_stream =
pair_read.read_pk_sorted_for_diff_with_type(&before,
&core_options, &diff_read_type)?;
let after_stream =
@@ -621,11 +654,13 @@ impl<'a> PaimonTableRead<'a> {
let before = before.to_vec();
let after = after.to_vec();
let parquet_read_budget = self.parquet_read_budget()?;
+ let blob_parallelism = self.blob_parallelism;
Ok(Box::pin(async_stream::try_stream! {
let core_options = CoreOptions::new(table.schema().options());
let pair_read = PaimonTableRead::new(&table,
diff_read_type.clone(), data_predicates)
- .with_parquet_read_budget(parquet_read_budget);
+ .with_parquet_read_budget(parquet_read_budget)
+ .with_blob_parallelism(blob_parallelism);
let before_stream = pair_read.read_pk_sorted_for_diff_with_type(
&before,
&core_options,
@@ -894,6 +929,7 @@ impl<'a> PaimonTableRead<'a> {
self.table.rest_env().cloned(),
)?
.with_batch_size(Some(core_options.read_batch_size()?))
+ .with_blob_parallelism(self.blob_parallelism)
.with_parquet_read_budget(Some(self.parquet_read_budget()?))
.with_read_timing(self.data_file_read_timing.clone());
reader.read(data_splits)
@@ -916,6 +952,7 @@ impl<'a> PaimonTableRead<'a> {
)
.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)
.with_parquet_read_budget(Some(self.parquet_read_budget()?))
.with_read_timing(self.data_file_read_timing.clone());
// The engine decoder filter is safe only on the plain append/raw path.
