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 d842653f fix(arrow): enforce exact Mosaic predicate filtering (#694)
d842653f is described below
commit d842653ffef234043e10edfb06cd074479e4a1a8
Author: QuakeWang <[email protected]>
AuthorDate: Sun Aug 9 21:05:57 2026 +0800
fix(arrow): enforce exact Mosaic predicate filtering (#694)
---
.../integrations/datafusion/tests/mosaic_tables.rs | 38 +++++++++
crates/paimon/src/arrow/format/mod.rs | 7 +-
crates/paimon/src/arrow/format/mosaic.rs | 93 ++++++++++++++--------
crates/paimon/src/table/data_file_reader.rs | 6 +-
4 files changed, 102 insertions(+), 42 deletions(-)
diff --git a/crates/integrations/datafusion/tests/mosaic_tables.rs
b/crates/integrations/datafusion/tests/mosaic_tables.rs
index 7ada5e3c..37c2927d 100644
--- a/crates/integrations/datafusion/tests/mosaic_tables.rs
+++ b/crates/integrations/datafusion/tests/mosaic_tables.rs
@@ -24,6 +24,9 @@ use std::sync::Arc;
use datafusion::arrow::array::{Int32Array, Int64Array};
use datafusion::arrow::record_batch::RecordBatch;
+use futures::TryStreamExt;
+use paimon::catalog::Identifier;
+use paimon::spec::{Datum, PredicateBuilder};
use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options};
use paimon_datafusion::SQLContext;
@@ -183,3 +186,38 @@ async fn test_read_pypaimon_mosaic_fixture() {
);
assert_eq!(filtered_rows, vec![(2, "Bob".to_string(), 20)]);
}
+
+#[tokio::test]
+async fn test_direct_core_read_applies_mosaic_filter_exactly() {
+ let (_tmp, warehouse) = extract_test_warehouse();
+ let mut options = Options::new();
+ options.set(CatalogOptions::WAREHOUSE, warehouse);
+ let catalog = FileSystemCatalog::new(options).expect("Failed to create
catalog");
+ let table = catalog
+ .get_table(&Identifier::new("default", FIXTURE_TABLE))
+ .await
+ .expect("Failed to load Mosaic fixture table");
+
+ let plan = table
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .expect("Failed to plan Mosaic fixture scan");
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .equal("id", Datum::Int(2))
+ .expect("Failed to build id predicate");
+ let read = paimon::table::TableRead::new(&table,
table.schema().fields().to_vec(), Vec::new())
+ .with_filter(predicate);
+ let batches = read
+ .to_arrow(plan.splits())
+ .expect("Failed to create direct core stream")
+ .try_collect::<Vec<_>>()
+ .await
+ .expect("Failed to read Mosaic fixture directly");
+
+ assert_eq!(
+ collect_id_name_score(&batches),
+ vec![(2, "Bob".to_string(), 20)]
+ );
+}
diff --git a/crates/paimon/src/arrow/format/mod.rs
b/crates/paimon/src/arrow/format/mod.rs
index 81f1876e..7a5abbd8 100644
--- a/crates/paimon/src/arrow/format/mod.rs
+++ b/crates/paimon/src/arrow/format/mod.rs
@@ -68,12 +68,11 @@ pub(crate) trait FormatFileReader: Send + Sync {
/// requested output by name, so extra columns are harmless.
///
/// Predicate exactness is per-format, NOT a blanket guarantee:
- /// - Parquet, ORC, Avro, Row, and Vortex apply the predicate **exactly** —
+ /// - Parquet, ORC, Avro, Row, Mosaic, and Vortex apply the predicate
**exactly** —
/// each emitted batch contains only rows matching the pushed-down
predicate
/// (native pushdown for pruning + a row-level residual pass for the
rest).
- /// - Blob does not evaluate predicates at all; Mosaic applies only
- /// stats-level (row-group) pruning. For those, non-matching rows may
- /// survive and the caller must not assume exactness.
+ /// - Blob does not evaluate predicates at all. Non-matching rows may
survive,
+ /// and the caller must not assume exactness.
/// `row_selection` is a pre-merged list of 0-based inclusive row ranges
/// (DV + row_ranges already combined by the caller).
async fn read_batch_stream(
diff --git a/crates/paimon/src/arrow/format/mosaic.rs
b/crates/paimon/src/arrow/format/mosaic.rs
index bde72203..356fcc49 100644
--- a/crates/paimon/src/arrow/format/mosaic.rs
+++ b/crates/paimon/src/arrow/format/mosaic.rs
@@ -17,7 +17,10 @@
use super::{FilePredicates, FormatFileReader};
use crate::arrow::build_target_arrow_schema;
-use crate::arrow::filtering::{predicates_may_match_with_schema, StatsAccessor};
+use crate::arrow::filtering::{
+ predicates_may_match_with_schema, remap_predicates_to_file, StatsAccessor,
+};
+use crate::arrow::residual::{filter_record_batch_by_predicates,
widen_scan_fields};
use crate::io::FileRead;
use crate::spec::{DataField, DataType as PaimonDataType, Datum, Predicate};
use crate::table::{ArrowRecordBatchStream, RowRange};
@@ -129,26 +132,36 @@ fn read_mosaic_batches_blocking(
.iter()
.map(|column| column.name.as_str())
.collect::<HashSet<_>>();
- let existing_read_fields = read_fields
+ // Residual filtering needs every predicate column, including columns that
+ // are not part of the requested projection. DataFileReader drops extras by
+ // name after this format reader returns.
+ let scan_fields = widen_scan_fields(&read_fields, predicates.as_ref());
+ let existing_scan_fields = scan_fields
.iter()
.filter(|field| file_column_names.contains(field.name()))
.cloned()
.collect::<Vec<_>>();
- let read_schema = build_target_arrow_schema(&existing_read_fields)?;
+ // A Mosaic file may physically omit columns still declared by its logical
+ // schema. Remap against the columns actually scanned so missing predicate
+ // columns use the existing all-NULL semantics before stats and residuals.
+ let predicates = predicates.map(|predicates| FilePredicates {
+ predicates: remap_predicates_to_file(
+ &predicates.predicates,
+ &predicates.file_fields,
+ &existing_scan_fields,
+ ),
+ row_filter_factory: None,
+ file_fields: existing_scan_fields.clone(),
+ });
+ let read_schema = build_target_arrow_schema(&existing_scan_fields)?;
validate_mosaic_schema(&read_schema)?;
- let projected_names = existing_read_fields
+ let projected_names = existing_scan_fields
.iter()
.map(|field| field.name().to_string())
.collect::<Vec<_>>();
- let all_projected_columns_missing = !read_fields.is_empty() &&
projected_names.is_empty();
- let predicate_state = predicates.map(|predicates| {
- let file_column_indices =
- build_file_column_indices(mosaic_reader.schema(),
&predicates.file_fields);
- (
- predicates.predicates,
- predicates.file_fields,
- file_column_indices,
- )
+ let all_scan_columns_missing = !scan_fields.is_empty() &&
projected_names.is_empty();
+ let predicate_column_indices = predicates.as_ref().map(|predicates| {
+ build_file_column_indices(mosaic_reader.schema(),
&predicates.file_fields)
});
let mut row_group_start = 0usize;
@@ -175,7 +188,9 @@ fn read_mosaic_batches_blocking(
}
}
- if let Some((predicates, file_fields, file_column_indices)) =
&predicate_state {
+ if let (Some(predicates), Some(file_column_indices)) =
+ (predicates.as_ref(), predicate_column_indices.as_ref())
+ {
let row_group_stats = mosaic_reader
.row_group_stats(row_group_index)
.map_err(mosaic_read_error)?;
@@ -183,14 +198,14 @@ fn read_mosaic_batches_blocking(
row_group_rows,
row_group_stats,
file_column_indices,
- predicates,
- file_fields,
+ &predicates.predicates,
+ &predicates.file_fields,
)? {
continue;
}
}
- let batch = if all_projected_columns_missing {
+ let batch = if all_scan_columns_missing {
let row_count = selected_slices
.as_ref()
.map_or(row_group_rows, |slices| selected_row_count(slices));
@@ -207,6 +222,12 @@ fn read_mosaic_batches_blocking(
let batch =
row_group_reader.read_columns().map_err(mosaic_read_error)?;
take_row_slices(batch, selected_slices.as_deref(), &read_schema)?
};
+ let batch = match predicates.as_ref() {
+ Some(predicates) => {
+ filter_record_batch_by_predicates(batch, predicates,
&existing_scan_fields)?
+ }
+ None => batch,
+ };
for chunk in split_batch(batch, batch_size) {
if !send_batch(chunk) {
return Ok(());
@@ -1112,7 +1133,7 @@ mod tests {
}
#[tokio::test]
- async fn test_read_predicate_prunes_non_matching_row_groups() {
+ async fn test_read_predicate_prunes_and_filters_row_groups() {
let fields = data_fields();
let builder = PredicateBuilder::new(&fields);
let predicates = predicate_file_predicates(
@@ -1124,7 +1145,7 @@ mod tests {
.await
.unwrap();
- assert_eq!(collect_i32_column(&batches, 0), vec![10, 11]);
+ assert_eq!(collect_i32_column(&batches, 0), vec![10]);
}
#[tokio::test]
@@ -1144,7 +1165,7 @@ mod tests {
}
#[tokio::test]
- async fn test_read_predicate_missing_stats_fails_open() {
+ async fn test_read_predicate_missing_stats_still_filters_rows() {
let fields = data_fields();
let builder = PredicateBuilder::new(&fields);
let predicates = predicate_file_predicates(
@@ -1156,7 +1177,7 @@ mod tests {
.await
.unwrap();
- assert_eq!(collect_i32_column(&batches, 0), vec![1, 2, 10, 11, 20,
21]);
+ assert!(collect_i32_column(&batches, 0).is_empty());
}
#[tokio::test]
@@ -1194,7 +1215,7 @@ mod tests {
.await
.unwrap();
- assert_eq!(collect_i32_column(&batches, 0), vec![30, 40]);
+ assert_eq!(collect_i32_column(&batches, 0), vec![30]);
}
#[tokio::test]
@@ -1244,13 +1265,15 @@ mod tests {
}
#[tokio::test]
- async fn test_read_predicate_missing_column_false_prunes_all_row_groups() {
- let fields = data_fields();
- let predicates = predicate_file_predicates(fields.clone(),
vec![Predicate::AlwaysFalse]);
- let projected = vec![
- fields[0].clone(),
- field(3, "new_score", DataType::Int(IntType::with_nullable(true))),
- ];
+ async fn
test_read_predicate_missing_column_comparison_prunes_all_row_groups() {
+ let mut fields = data_fields();
+ let missing_field = field(3, "new_score",
DataType::Int(IntType::with_nullable(true)));
+ fields.push(missing_field.clone());
+ let predicate = PredicateBuilder::new(&fields)
+ .equal("new_score", Datum::Int(1))
+ .unwrap();
+ let predicates = predicate_file_predicates(fields.clone(),
vec![predicate]);
+ let projected = vec![fields[0].clone(), missing_field];
let data = multi_row_group_mosaic(vec!["id".to_string()]);
let batches = read_batches_with_predicates(data, &projected,
Some(&predicates), None)
.await
@@ -1261,12 +1284,12 @@ mod tests {
#[tokio::test]
async fn test_read_predicate_missing_column_is_null_keeps_row_groups() {
- let fields = data_fields();
- let predicates = predicate_file_predicates(fields.clone(),
vec![Predicate::AlwaysTrue]);
- let projected = vec![
- fields[0].clone(),
- field(3, "new_score", DataType::Int(IntType::with_nullable(true))),
- ];
+ let mut fields = data_fields();
+ let missing_field = field(3, "new_score",
DataType::Int(IntType::with_nullable(true)));
+ fields.push(missing_field.clone());
+ let predicate =
PredicateBuilder::new(&fields).is_null("new_score").unwrap();
+ let predicates = predicate_file_predicates(fields.clone(),
vec![predicate]);
+ let projected = vec![fields[0].clone(), missing_field];
let data = multi_row_group_mosaic(vec!["id".to_string()]);
let batches = read_batches_with_predicates(data, &projected,
Some(&predicates), None)
.await
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 591c88e9..5954984a 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -1804,8 +1804,8 @@ mod tests {
assert_eq!(ids, vec![1, 2, 3, 4]);
}
- /// Row-group predicate pruning, deletion vectors and projection must
compose correctly:
- /// the predicate keeps one row group, the DV deletes one of its rows,
projection keeps `id`.
+ /// Exact predicate filtering, deletion vectors and projection must
compose correctly:
+ /// the predicate selects `id = 10`, the DV deletes that row, and no rows
remain.
#[tokio::test]
async fn test_mosaic_predicate_dv_projection_combination() {
let fields = pk_fields();
@@ -1869,7 +1869,7 @@ mod tests {
.unwrap();
assert_eq!(batches.iter().map(|b| b.num_columns()).max(), Some(1));
- assert_eq!(collect_ids(&batches), vec![11]);
+ assert!(collect_ids(&batches).is_empty());
}
}
