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 ede9ceb7 [rust] Add parquet.filter.columnindex.enabled option to
control Page Index pruning (#875)
ede9ceb7 is described below
commit ede9ceb7af797f09f8d57b2a96efb8dc23040443
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Sep 20 10:22:10 2026 +0800
[rust] Add parquet.filter.columnindex.enabled option to control Page Index
pruning (#875)
---
crates/paimon/src/arrow/format/mod.rs | 13 +-
crates/paimon/src/arrow/format/parquet.rs | 179 +++++++++++++++++++--
crates/paimon/src/spec/core_options.rs | 49 ++++++
crates/paimon/src/spec/schema.rs | 21 +++
crates/paimon/src/table/audit_log_table/read.rs | 1 +
crates/paimon/src/table/data_evolution_reader.rs | 16 ++
crates/paimon/src/table/data_file_reader.rs | 17 +-
crates/paimon/src/table/format_table_read.rs | 2 +
.../paimon/src/table/full_text_search_builder.rs | 3 +-
crates/paimon/src/table/hybrid_search_builder.rs | 6 +-
crates/paimon/src/table/kv_file_reader.rs | 1 +
crates/paimon/src/table/pk_vector_read.rs | 9 +-
crates/paimon/src/table/table_read.rs | 3 +
crates/paimon/src/table/vector_search_result.rs | 6 +-
docs/src/getting-started.md | 7 +
15 files changed, 301 insertions(+), 32 deletions(-)
diff --git a/crates/paimon/src/arrow/format/mod.rs
b/crates/paimon/src/arrow/format/mod.rs
index 4fc9f93b..4cd42a3c 100644
--- a/crates/paimon/src/arrow/format/mod.rs
+++ b/crates/paimon/src/arrow/format/mod.rs
@@ -195,28 +195,29 @@ pub(crate) fn create_format_reader(
path,
blob_as_descriptor,
read_fields,
+ &HashMap::new(),
None,
blob::DEFAULT_BLOB_READ_PARALLELISM,
MosaicPrefetchOptions::default(),
)
}
-/// Create a format reader with a scan-shared Parquet resource budget and the
-/// per-file Mosaic prefetch settings.
+/// Create a format reader with table options and runtime read resources.
pub(crate) fn create_format_reader_with_budget(
path: &str,
blob_as_descriptor: bool,
read_fields: &[DataField],
+ table_options: &HashMap<String, String>,
parquet_read_budget: Option<Arc<ReadBudget>>,
blob_parallelism: usize,
mosaic_prefetch: MosaicPrefetchOptions,
) -> crate::Result<Box<dyn FormatFileReader>> {
let lower = path.to_ascii_lowercase();
let reader: Box<dyn FormatFileReader> = if lower.ends_with(".parquet") {
- Box::new(match parquet_read_budget {
- Some(read_budget) =>
parquet::ParquetFormatReader::with_read_budget(read_budget),
- None => parquet::ParquetFormatReader::default(),
- })
+ Box::new(parquet::ParquetFormatReader::with_options(
+ table_options,
+ parquet_read_budget,
+ )?)
} else if lower.ends_with(".blob") {
Box::new(
blob::BlobFormatReader::new(path.to_string(), blob_as_descriptor)
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index 758fb36e..d4ddd584 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -54,17 +54,45 @@ use std::ops::Range;
use std::sync::Arc;
use tokio::sync::mpsc;
-#[derive(Default)]
pub(crate) struct ParquetFormatReader {
read_budget: Option<Arc<ReadBudget>>,
+ page_index_enabled: bool,
+}
+
+impl Default for ParquetFormatReader {
+ fn default() -> Self {
+ Self {
+ read_budget: None,
+ page_index_enabled: true,
+ }
+ }
}
impl ParquetFormatReader {
- pub(crate) fn with_read_budget(read_budget: Arc<ReadBudget>) -> Self {
+ pub(crate) fn with_options(
+ options: &HashMap<String, String>,
+ read_budget: Option<Arc<ReadBudget>>,
+ ) -> crate::Result<Self> {
+ Ok(Self {
+ read_budget,
+ page_index_enabled: crate::spec::CoreOptions::new(options)
+ .parquet_filter_column_index_enabled()?,
+ })
+ }
+
+ #[cfg(test)]
+ fn with_read_budget(read_budget: Arc<ReadBudget>) -> Self {
Self {
read_budget: Some(read_budget),
+ ..Default::default()
}
}
+
+ #[cfg(test)]
+ fn with_page_index_enabled(mut self, enabled: bool) -> Self {
+ self.page_index_enabled = enabled;
+ self
+ }
}
enum ParquetRowGroupMessage {
@@ -375,11 +403,13 @@ impl FormatFileReader for ParquetFormatReader {
// Predicates need both indexes for page-stat pruning. Row selection
only
// needs OffsetIndex so arrow-rs can avoid fetching unselected pages.
let mut arrow_options = ArrowReaderOptions::new();
- if !preds.is_empty() {
- arrow_options =
arrow_options.with_column_index_policy(PageIndexPolicy::Optional);
- }
- if !preds.is_empty() || row_selection.is_some() {
- arrow_options =
arrow_options.with_offset_index_policy(PageIndexPolicy::Optional);
+ if self.page_index_enabled {
+ if !preds.is_empty() {
+ arrow_options =
arrow_options.with_column_index_policy(PageIndexPolicy::Optional);
+ }
+ if !preds.is_empty() || row_selection.is_some() {
+ arrow_options =
arrow_options.with_offset_index_policy(PageIndexPolicy::Optional);
+ }
}
let mut batch_stream_builder =
ParquetRecordBatchStreamBuilder::new_with_options(arrow_file_reader,
arrow_options)
@@ -497,8 +527,11 @@ impl FormatFileReader for ParquetFormatReader {
// Page-level selection. Returns `None` when ColumnIndex / OffsetIndex
are
// absent (page index not loaded, older files, writer without page
index)
// or when no page could be skipped, so intersecting is a no-op then.
- let page_selection =
- build_predicate_page_selection(batch_stream_builder.metadata(),
preds, file_fields)?;
+ let page_selection = if self.page_index_enabled {
+ build_predicate_page_selection(batch_stream_builder.metadata(),
preds, file_fields)?
+ } else {
+ None
+ };
combined_selection =
intersect_optional_row_selections(combined_selection, page_selection);
if let Some(ref ranges) = row_selection {
@@ -3552,6 +3585,99 @@ mod tests {
buf
}
+ async fn write_page_pruning_io_parquet() -> Vec<u8> {
+ const ROWS: i32 = 1024;
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("payload", ArrowDataType::Utf8, false),
+ ]));
+ let payloads = (0..ROWS)
+ .map(|row| format!("{row:04}-{}", "x".repeat(4096)))
+ .collect::<Vec<_>>();
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from_iter_values(0..ROWS)),
+ Arc::new(StringArray::from(payloads)),
+ ],
+ )
+ .unwrap();
+ let props = parquet::file::properties::WriterProperties::builder()
+ .set_compression(Compression::UNCOMPRESSED)
+ .set_dictionary_enabled(false)
+ .set_data_page_row_count_limit(64)
+ .set_write_batch_size(64)
+ .set_max_row_group_row_count(Some(ROWS as usize))
+ .build();
+ let mut buf = Vec::new();
+ let mut writer = AsyncArrowWriter::try_new(&mut buf, schema,
Some(props)).unwrap();
+ writer.write(&batch).await.unwrap();
+ writer.close().await.unwrap();
+ buf
+ }
+
+ async fn read_page_pruning_io(data: Bytes, page_index_enabled: bool) ->
(Vec<String>, u64) {
+ let file_size = data.len() as u64;
+ let file_read = TrackingFileRead::new(data);
+ let tracker = file_read.clone();
+ let fields = vec![
+ int_field("id"),
+ DataField::new(
+ 1,
+ "payload".to_string(),
+ DataType::VarChar(VarCharType::string_type()),
+ ),
+ ];
+ let predicates = FilePredicates {
+ predicates: vec![id_leaf(PredicateOperator::Eq,
vec![Datum::Int(1)])],
+ row_filter_factory: None,
+ file_fields: fields.clone(),
+ };
+ let stream = ParquetFormatReader::default()
+ .with_page_index_enabled(page_index_enabled)
+ .read_batch_stream(
+ Box::new(file_read),
+ file_size,
+ &fields[1..],
+ Some(&predicates),
+ Some(64),
+ None,
+ )
+ .await
+ .unwrap();
+ tracker.reset();
+ let batches = stream.try_collect::<Vec<_>>().await.unwrap();
+ let payloads = batches
+ .iter()
+ .flat_map(|batch| {
+ let values = batch
+ .column_by_name("payload")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ (0..values.len())
+ .map(|index| values.value(index).to_string())
+ .collect::<Vec<_>>()
+ })
+ .collect();
+ (payloads, tracker.bytes_read())
+ }
+
+ #[tokio::test]
+ async fn test_predicate_page_pruning_option_controls_io_not_results() {
+ let data = Bytes::from(write_page_pruning_io_parquet().await);
+ let (enabled_rows, enabled_bytes) = read_page_pruning_io(data.clone(),
true).await;
+ let (disabled_rows, disabled_bytes) = read_page_pruning_io(data,
false).await;
+
+ assert_eq!(enabled_rows, disabled_rows);
+ assert_eq!(enabled_rows.len(), 1);
+ assert!(
+ enabled_bytes * 2 < disabled_bytes,
+ "page-index read used {enabled_bytes} bytes; disabled read used
{disabled_bytes} bytes"
+ );
+ }
+
#[derive(Clone)]
struct TrackingFileRead {
data: Bytes,
@@ -3625,7 +3751,11 @@ mod tests {
buf
}
- async fn read_nested_rows(data: Bytes, row_selection:
Option<Vec<RowRange>>) -> (usize, u64) {
+ async fn read_nested_rows(
+ data: Bytes,
+ row_selection: Option<Vec<RowRange>>,
+ page_index_enabled: Option<bool>,
+ ) -> (usize, u64) {
let file_size = data.len() as u64;
let file_read = TrackingFileRead::new(data);
let tracker = file_read.clone();
@@ -3634,7 +3764,11 @@ mod tests {
"items".to_string(),
DataType::Array(ArrayType::new(DataType::Int(IntType::new()))),
)];
- let stream = ParquetFormatReader::default()
+ let reader = match page_index_enabled {
+ Some(enabled) =>
ParquetFormatReader::default().with_page_index_enabled(enabled),
+ None => ParquetFormatReader::default(),
+ };
+ let stream = reader
.read_batch_stream(
Box::new(file_read),
file_size,
@@ -3659,15 +3793,26 @@ mod tests {
#[tokio::test]
async fn test_row_selection_prunes_nested_page_io() {
let data = Bytes::from(write_nested_multi_page_parquet().await);
- let (all_rows, all_bytes) = read_nested_rows(data.clone(), None).await;
- let (selected_rows, selected_bytes) =
- read_nested_rows(data, Some(vec![RowRange::new(0, 0)])).await;
+ let (all_rows, all_bytes) = read_nested_rows(data.clone(), None,
None).await;
+ let selection = Some(vec![RowRange::new(0, 0)]);
+ let (default_rows, default_bytes) =
+ read_nested_rows(data.clone(), selection.clone(), None).await;
+ let (enabled_rows, enabled_bytes) =
+ read_nested_rows(data.clone(), selection.clone(),
Some(true)).await;
+ let (disabled_rows, disabled_bytes) = read_nested_rows(data,
selection, Some(false)).await;
assert_eq!(all_rows, 1024);
- assert_eq!(selected_rows, 1);
+ assert_eq!(default_rows, 1);
+ assert_eq!(enabled_rows, 1);
+ assert_eq!(disabled_rows, 1);
+ assert_eq!(default_bytes, enabled_bytes);
+ assert!(
+ enabled_bytes * 2 < disabled_bytes,
+ "page-index read used {enabled_bytes} bytes; disabled read used
{disabled_bytes} bytes"
+ );
assert!(
- selected_bytes * 2 < all_bytes,
- "selected read used {selected_bytes} bytes; full read used
{all_bytes} bytes"
+ enabled_bytes * 2 < all_bytes,
+ "selected read used {enabled_bytes} bytes; full read used
{all_bytes} bytes"
);
}
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index a741a087..58131b6c 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -92,6 +92,7 @@ const MANIFEST_SIDECAR_ENABLED_OPTION: &str =
"manifest.sidecar.enabled";
const MANIFEST_SORT_ENABLED_OPTION: &str = "manifest-sort.enabled";
const WRITE_PARQUET_BUFFER_SIZE_OPTION: &str = "write.parquet-buffer-size";
const READ_BATCH_SIZE_OPTION: &str = "read.batch-size";
+const PARQUET_FILTER_COLUMN_INDEX_ENABLED_OPTION: &str =
"parquet.filter.columnindex.enabled";
const PARQUET_ROW_GROUP_PARALLELISM_OPTION: &str =
"read.parquet.row-group.parallelism";
pub(crate) const PARQUET_ROW_GROUP_MAX_INFLIGHT_BYTES_OPTION: &str =
"read.parquet.row-group.max-inflight-bytes";
@@ -733,6 +734,22 @@ impl<'a> CoreOptions<'a> {
.unwrap_or(true)
}
+ /// Whether Parquet page indexes are used for page pruning. Default is
true.
+ pub fn parquet_filter_column_index_enabled(&self) -> crate::Result<bool> {
+ let Some(raw) =
self.options.get(PARQUET_FILTER_COLUMN_INDEX_ENABLED_OPTION) else {
+ return Ok(true);
+ };
+ match raw.trim().to_ascii_lowercase().as_str() {
+ "true" => Ok(true),
+ "false" => Ok(false),
+ _ => Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Option '{PARQUET_FILTER_COLUMN_INDEX_ENABLED_OPTION}'
must be true or false, got: {raw}"
+ ),
+ }),
+ }
+ }
+
/// The declared [`TableType`], defaulting to [`TableType::Table`].
/// Fails on a value this client does not know.
pub fn table_type(&self) -> crate::Result<TableType> {
@@ -1788,6 +1805,38 @@ mod tests {
}
}
+ #[test]
+ fn test_parquet_filter_column_index_enabled() {
+ let options = HashMap::new();
+ assert!(CoreOptions::new(&options)
+ .parquet_filter_column_index_enabled()
+ .unwrap());
+
+ for (value, expected) in [("true", true), ("false", false), (" FALSE
", false)] {
+ let options = HashMap::from([(
+ PARQUET_FILTER_COLUMN_INDEX_ENABLED_OPTION.to_string(),
+ value.to_string(),
+ )]);
+ assert_eq!(
+ CoreOptions::new(&options)
+ .parquet_filter_column_index_enabled()
+ .unwrap(),
+ expected
+ );
+ }
+
+ let options = HashMap::from([(
+ PARQUET_FILTER_COLUMN_INDEX_ENABLED_OPTION.to_string(),
+ "invalid".to_string(),
+ )]);
+ let error = CoreOptions::new(&options)
+ .parquet_filter_column_index_enabled()
+ .unwrap_err();
+ assert!(error
+ .to_string()
+ .contains("parquet.filter.columnindex.enabled"));
+ }
+
#[test]
fn test_source_split_defaults() {
let options = HashMap::new();
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 662889ad..11dc4365 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -2520,6 +2520,27 @@ mod tests {
assert!(copied.core_options().index_file_in_data_file_dir());
}
+ #[test]
+ fn a_dynamic_copy_can_disable_parquet_page_indexes() {
+ let stored = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .option("parquet.filter.columnindex.enabled", "true")
+ .build()
+ .unwrap(),
+ );
+ let copied = stored.copy_with_options(HashMap::from([(
+ "parquet.filter.columnindex.enabled".to_string(),
+ "false".to_string(),
+ )]));
+
+ assert!(!copied
+ .core_options()
+ .parquet_filter_column_index_enabled()
+ .unwrap());
+ }
+
#[test]
fn test_copy_with_replaced_options() {
let schema = Schema::builder()
diff --git a/crates/paimon/src/table/audit_log_table/read.rs
b/crates/paimon/src/table/audit_log_table/read.rs
index 5b4aac43..175ce60e 100644
--- a/crates/paimon/src/table/audit_log_table/read.rs
+++ b/crates/paimon/src/table/audit_log_table/read.rs
@@ -104,6 +104,7 @@ impl<'a> AuditLogRead<'a> {
.with_file_index_read_enabled(core_options.file_index_read_enabled())
.with_batch_size(Some(core_options.read_batch_size()?))
.with_parquet_read_budget(Some(Arc::clone(&parquet_read_budget)))
+ .with_table_options(self.read.table.schema().options().clone())
.read(&raw_splits)?;
let merge_stream = KeyValueFileReader::new(
self.read.table.file_io.clone(),
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index ce0ece4d..ddd80975 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -117,6 +117,7 @@ pub(crate) struct DataEvolutionReader {
blob_parallelism: usize,
batch_size: Option<usize>,
parquet_read_budget: Option<Arc<ReadBudget>>,
+ table_options: Arc<HashMap<String, String>>,
mosaic_prefetch: MosaicPrefetchOptions,
read_timing: Option<Arc<DataFileReadTiming>>,
}
@@ -197,6 +198,7 @@ impl DataEvolutionReader {
blob_parallelism: DEFAULT_BLOB_READ_PARALLELISM,
batch_size: None,
parquet_read_budget: None,
+ table_options: Arc::new(HashMap::new()),
mosaic_prefetch: MosaicPrefetchOptions::default(),
read_timing: None,
})
@@ -222,6 +224,14 @@ impl DataEvolutionReader {
self
}
+ pub(crate) fn with_table_options(
+ mut self,
+ options: impl Into<Arc<HashMap<String, String>>>,
+ ) -> Self {
+ self.table_options = options.into();
+ self
+ }
+
pub(crate) fn with_mosaic_prefetch(mut self, mosaic_prefetch:
MosaicPrefetchOptions) -> Self {
self.mosaic_prefetch = mosaic_prefetch;
self
@@ -275,6 +285,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_table_options(Arc::clone(&self.table_options))
.with_mosaic_prefetch(self.mosaic_prefetch)
.with_read_timing(self.read_timing.clone());
@@ -576,6 +587,7 @@ impl DataEvolutionReader {
)?
.with_batch_size(self.batch_size)
.with_parquet_read_budget(self.parquet_read_budget.clone())
+ .with_table_options(Arc::clone(&self.table_options))
.with_mosaic_prefetch(self.mosaic_prefetch);
let mut stream = prescan.read(splits)?;
let mut view_structs = HashSet::new();
@@ -641,6 +653,7 @@ impl DataEvolutionReader {
let blob_parallelism = self.blob_parallelism;
let batch_size = self.batch_size;
let parquet_read_budget = self.parquet_read_budget.clone();
+ let table_options = Arc::clone(&self.table_options);
let mosaic_prefetch = self.mosaic_prefetch;
let read_timing = self.read_timing.clone();
let anchor_deletion_vector = anchor_deletion_vector.clone();
@@ -730,6 +743,7 @@ impl DataEvolutionReader {
blob_as_descriptor,
blob_parallelism,
source_parquet_read_budget.clone(),
+ Arc::clone(&table_options),
mosaic_prefetch,
read_timing.clone(),
anchor_deletion_vector.as_ref(),
@@ -1271,6 +1285,7 @@ fn open_source_stream(
blob_as_descriptor: bool,
blob_parallelism: usize,
parquet_read_budget: Option<Arc<ReadBudget>>,
+ table_options: Arc<HashMap<String, String>>,
mosaic_prefetch: MosaicPrefetchOptions,
read_timing: Option<Arc<DataFileReadTiming>>,
anchor_deletion_vector: Option<&DeletionVectorContext>,
@@ -1341,6 +1356,7 @@ fn open_source_stream(
.with_blob_as_descriptor(blob_as_descriptor)
.with_blob_parallelism(blob_parallelism)
.with_parquet_read_budget(parquet_read_budget)
+ .with_table_options(table_options)
.with_mosaic_prefetch(mosaic_prefetch)
.with_read_timing(read_timing);
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 7019e9bc..d72703b0 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -36,6 +36,7 @@ use arrow_array::{Array, Int64Array, RecordBatch};
use async_stream::try_stream;
use futures::StreamExt;
use roaring::RoaringBitmap;
+use std::collections::HashMap;
use std::ops::Range;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
@@ -121,6 +122,7 @@ pub(crate) struct DataFileReader {
blob_parallelism: usize,
batch_size: Option<usize>,
parquet_read_budget: Option<Arc<ReadBudget>>,
+ table_options: Arc<HashMap<String, String>>,
mosaic_prefetch: MosaicPrefetchOptions,
read_timing: Option<Arc<DataFileReadTiming>>,
}
@@ -147,6 +149,7 @@ impl DataFileReader {
blob_parallelism: DEFAULT_BLOB_READ_PARALLELISM,
batch_size: None,
parquet_read_budget: None,
+ table_options: Arc::new(HashMap::new()),
mosaic_prefetch: MosaicPrefetchOptions::default(),
read_timing: None,
}
@@ -181,6 +184,14 @@ impl DataFileReader {
self
}
+ pub(crate) fn with_table_options(
+ mut self,
+ options: impl Into<Arc<HashMap<String, String>>>,
+ ) -> Self {
+ self.table_options = options.into();
+ self
+ }
+
pub(crate) fn with_mosaic_prefetch(mut self, mosaic_prefetch:
MosaicPrefetchOptions) -> Self {
self.mosaic_prefetch = mosaic_prefetch;
self
@@ -463,10 +474,11 @@ impl DataFileReader {
.flatten();
let file_io = self.file_io.clone();
let split = split.clone();
+ let batch_size = self.batch_size;
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 table_options = Arc::clone(&self.table_options);
let mosaic_prefetch = self.mosaic_prefetch;
let read_timing = self.read_timing.clone();
@@ -526,6 +538,7 @@ impl DataFileReader {
&path_to_read,
blob_as_descriptor,
&format_read_fields,
+ &table_options,
parquet_read_budget,
blob_parallelism,
mosaic_prefetch,
@@ -738,6 +751,7 @@ impl DataFileReader {
let blob_as_descriptor = self.blob_as_descriptor;
let blob_parallelism = self.blob_parallelism;
let parquet_read_budget = self.parquet_read_budget.clone();
+ let table_options = Arc::clone(&self.table_options);
let mosaic_prefetch = self.mosaic_prefetch;
let target_schema = build_target_arrow_schema(&read_type)?;
@@ -802,6 +816,7 @@ impl DataFileReader {
&path_to_read,
blob_as_descriptor,
&format_read_fields,
+ &table_options,
parquet_read_budget,
blob_parallelism,
mosaic_prefetch,
diff --git a/crates/paimon/src/table/format_table_read.rs
b/crates/paimon/src/table/format_table_read.rs
index d2281405..e419e78e 100644
--- a/crates/paimon/src/table/format_table_read.rs
+++ b/crates/paimon/src/table/format_table_read.rs
@@ -130,6 +130,7 @@ impl<'a> FormatTableRead<'a> {
let mosaic_prefetch = configured_mosaic_prefetch(self.table)?;
let row_filter_factory = self.row_filter_factory.clone();
let parquet_read_budget = Some(self.parquet_read_budget()?);
+ let table_options = self.table.schema().options().clone();
let blob_parallelism = self.blob_parallelism;
Ok(try_stream! {
@@ -149,6 +150,7 @@ impl<'a> FormatTableRead<'a> {
.with_batch_size(batch_size)
.with_blob_parallelism(blob_parallelism)
.with_parquet_read_budget(parquet_read_budget.clone())
+ .with_table_options(table_options.clone())
.with_mosaic_prefetch(mosaic_prefetch);
if let Some(factory) = &row_filter_factory {
reader =
reader.with_row_filter_factory(Arc::clone(factory));
diff --git a/crates/paimon/src/table/full_text_search_builder.rs
b/crates/paimon/src/table/full_text_search_builder.rs
index fac4db4a..1c567162 100644
--- a/crates/paimon/src/table/full_text_search_builder.rs
+++ b/crates/paimon/src/table/full_text_search_builder.rs
@@ -273,7 +273,8 @@ impl<'a> FullTextSearchBuilder<'a> {
self.table.schema().fields().to_vec(),
self.table.schema().fields().to_vec(),
Vec::new(),
- );
+ )
+ .with_table_options(self.table.schema().options().clone());
let read = PrimaryKeyFullTextRead::new(
self.table.file_io().clone(),
materialize_reader,
diff --git a/crates/paimon/src/table/hybrid_search_builder.rs
b/crates/paimon/src/table/hybrid_search_builder.rs
index 795978b7..f6be569f 100644
--- a/crates/paimon/src/table/hybrid_search_builder.rs
+++ b/crates/paimon/src/table/hybrid_search_builder.rs
@@ -492,7 +492,8 @@ impl<'a> HybridSearchBuilder<'a> {
self.table.schema().fields().to_vec(),
self.table.schema().fields().to_vec(),
Vec::new(),
- );
+ )
+ .with_table_options(self.table.schema().options().clone());
let mut batches: Vec<RecordBatch> = Vec::new();
let mut ranked: Vec<RankedRow> = Vec::new();
@@ -634,7 +635,8 @@ impl<'a> HybridSearchBuilder<'a> {
table.schema().fields().to_vec(),
table.schema().fields().to_vec(),
Vec::new(),
- );
+ )
+ .with_table_options(table.schema().options().clone());
let read = PrimaryKeyFullTextRead::new(
table.file_io().clone(),
materialize_reader,
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index 0bcc97cd..9b2a868f 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -592,6 +592,7 @@ impl KeyValueFileReader {
)
.with_batch_size(Some(config.read_batch_size))
.with_parquet_read_budget(group_parquet_read_budget.clone())
+ .with_table_options(config.table_options.clone())
.with_mosaic_prefetch(config.mosaic_prefetch);
let run_schema_manager = config.schema_manager.clone();
let run_file_io = file_io.clone();
diff --git a/crates/paimon/src/table/pk_vector_read.rs
b/crates/paimon/src/table/pk_vector_read.rs
index 8a4ab204..59f74c44 100644
--- a/crates/paimon/src/table/pk_vector_read.rs
+++ b/crates/paimon/src/table/pk_vector_read.rs
@@ -300,7 +300,8 @@ async fn search_pk_raw_candidates_batch_with_plan(
table.schema().fields().to_vec(),
vec![vector_field.clone()],
Vec::new(),
- );
+ )
+ .with_table_options(table.schema().options().clone());
// Real ANN scorer + loader. Each segment source is opened lazily inside
its
// bucket leaf and dropped after scoring. Lumina keeps its buffered-byte
path;
@@ -447,7 +448,8 @@ async fn search_pk_raw_candidates_batch_with_plan(
table.schema().fields().to_vec(),
residual_read_type,
Vec::new(),
- );
+ )
+ .with_table_options(table.schema().options().clone());
let mut per_split = Vec::with_capacity(plan.splits.len());
for (index, split) in plan.splits.iter().enumerate() {
// The plan's selection for this split, so the residual is
@@ -611,7 +613,8 @@ async fn search_pk_candidates_batch_with_plan(
table.schema().fields().to_vec(),
vec![vector_field.clone()],
Vec::new(),
- );
+ )
+ .with_table_options(table.schema().options().clone());
rerank_indexed_positional(
&rerank_reader,
search.indexed,
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 47a0b0a7..d232614b 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -458,6 +458,7 @@ impl<'a> PaimonTableRead<'a> {
.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_table_options(self.table.schema().options().clone())
.with_mosaic_prefetch(configured_mosaic_prefetch(self.table)?);
let raw_stream = reader.read(&data_splits)?;
@@ -941,6 +942,7 @@ impl<'a> PaimonTableRead<'a> {
.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_table_options(self.table.schema().options().clone())
.with_mosaic_prefetch(configured_mosaic_prefetch(self.table)?)
.with_read_timing(self.data_file_read_timing.clone());
reader.read(data_splits)
@@ -965,6 +967,7 @@ impl<'a> PaimonTableRead<'a> {
.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_table_options(self.table.schema().options().clone())
.with_mosaic_prefetch(configured_mosaic_prefetch(self.table)?)
.with_read_timing(self.data_file_read_timing.clone());
// The engine decoder filter is safe only on the plain append/raw path.
diff --git a/crates/paimon/src/table/vector_search_result.rs
b/crates/paimon/src/table/vector_search_result.rs
index 5ceae40b..0a69ba45 100644
--- a/crates/paimon/src/table/vector_search_result.rs
+++ b/crates/paimon/src/table/vector_search_result.rs
@@ -201,7 +201,8 @@ impl SearchResultReadBuilder<'_> {
pub async fn read(&self) -> crate::Result<ArrowRecordBatchStream> {
let table = &self.result.table;
- CoreOptions::new(table.schema().options()).ensure_read_authorized()?;
+ let core_options = CoreOptions::new(table.schema().options());
+ core_options.ensure_read_authorized()?;
match &self.result.hits {
SearchHits::DataEvolution {
vector_column,
@@ -220,7 +221,8 @@ impl SearchResultReadBuilder<'_> {
table.schema().fields().to_vec(),
read_type,
Vec::new(),
- );
+ )
+ .with_table_options(table.schema().options().clone());
materialize_positions(positions, splits, &reader).await
}
}
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index c04e54b9..214d9d39 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -122,6 +122,13 @@ cross-backend rename is rejected.
with `with_provider` on the same builder. Without a provider, property-based
storage configuration behaves as before.
+## Parquet Page Pruning
+
+Parquet Page Index pruning is enabled by default. Set
+`parquet.filter.columnindex.enabled=false` to avoid fine-grained range
requests;
+this may reduce request count while increasing bytes read. The option does not
+change Page Index generation.
+
## Mosaic File Format
Mosaic data file reads are always available. The current Mosaic support is
read-only: Paimon Rust can read existing `.mosaic` data files, including array
and map columns, in a Paimon table, but it does not write Mosaic data files yet.
