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 f38096a0 [datafusion] Support runtime filters for Paimon scans (#549)
f38096a0 is described below
commit f38096a09c696cd2b5c9933ba44690244df9f23e
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Jul 20 14:28:35 2026 +0800
[datafusion] Support runtime filters for Paimon scans (#549)
---
benchmarks/tpcds/README.md | 5 +
benchmarks/tpcds/src/cli.rs | 4 +
benchmarks/tpcds/src/context.rs | 10 +-
benchmarks/tpcds/src/lib.rs | 9 +
benchmarks/tpcds/tests/smoke.rs | 1 +
crates/integrations/datafusion/src/config.rs | 61 ++
.../integrations/datafusion/src/filter_pushdown.rs | 2 +-
crates/integrations/datafusion/src/lib.rs | 1 +
.../datafusion/src/physical_plan/scan.rs | 823 ++++++++++++++++++++-
crates/integrations/datafusion/src/sql_context.rs | 55 +-
crates/integrations/datafusion/tests/blob_tests.rs | 3 +-
crates/paimon/src/arrow/format/avro.rs | 4 +
crates/paimon/src/arrow/format/mod.rs | 8 +-
crates/paimon/src/arrow/format/mosaic.rs | 2 +
crates/paimon/src/arrow/format/orc.rs | 2 +
crates/paimon/src/arrow/format/parquet.rs | 40 +-
crates/paimon/src/arrow/format/row.rs | 3 +
crates/paimon/src/arrow/format/vortex.rs | 11 +
crates/paimon/src/arrow/residual.rs | 7 +-
crates/paimon/src/table/data_evolution_reader.rs | 83 ++-
crates/paimon/src/table/data_file_reader.rs | 134 +++-
crates/paimon/src/table/format_table_read.rs | 10 +-
crates/paimon/src/table/kv_file_reader.rs | 65 +-
crates/paimon/src/table/table_read.rs | 48 +-
crates/paimon/src/table/vector_search_builder.rs | 2 +
docs/src/sql.md | 4 +-
26 files changed, 1324 insertions(+), 73 deletions(-)
diff --git a/benchmarks/tpcds/README.md b/benchmarks/tpcds/README.md
index f5872214..606f0909 100644
--- a/benchmarks/tpcds/README.md
+++ b/benchmarks/tpcds/README.md
@@ -164,6 +164,7 @@ target/release/paimon-tpcds-bench run \
--warmup 1 \
--iterations 3 \
--target-partitions 64 \
+ --parquet-pushdown-filters \
--memory-limit-gib 192 \
--spill-dir /nvme/datafusion-spill \
--max-spill-gib 1024
@@ -173,6 +174,10 @@ This is an end-to-end source comparison. Loading the data
into Paimon rewrites
the physical files, so it is not a pure measurement of catalog or manifest
overhead.
+`--parquet-pushdown-filters` only controls DataFusion's Parquet reader. Paimon
+always receives supported predicates for conservative pruning, while exact row
+filtering stays in the parent DataFusion operator for benchmark runs.
+
## Cache Protocol
Run and label cold and warm experiments separately:
diff --git a/benchmarks/tpcds/src/cli.rs b/benchmarks/tpcds/src/cli.rs
index 878819f2..ece7f9de 100644
--- a/benchmarks/tpcds/src/cli.rs
+++ b/benchmarks/tpcds/src/cli.rs
@@ -113,6 +113,9 @@ pub struct RuntimeArgs {
/// DataFusion execution partitions. Defaults to available CPUs.
#[arg(long)]
pub target_partitions: Option<usize>,
+ /// Evaluate pushed filters during Parquet scans, in addition to
statistics pruning.
+ #[arg(long)]
+ pub parquet_pushdown_filters: bool,
/// DataFusion memory limit in GiB. Omit for an unbounded pool.
#[arg(long)]
pub memory_limit_gib: Option<u64>,
@@ -132,6 +135,7 @@ impl RuntimeArgs {
.target_partitions
.unwrap_or(defaults.target_partitions)
.max(1),
+ parquet_pushdown_filters: self.parquet_pushdown_filters,
memory_limit_bytes:
self.memory_limit_gib.map(gib_to_usize).transpose()?,
spill_dir: self.spill_dir.clone(),
max_spill_bytes: self.max_spill_gib.map(gib_to_u64).transpose()?,
diff --git a/benchmarks/tpcds/src/context.rs b/benchmarks/tpcds/src/context.rs
index a4a4cd4e..c9b0d703 100644
--- a/benchmarks/tpcds/src/context.rs
+++ b/benchmarks/tpcds/src/context.rs
@@ -28,6 +28,8 @@ use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BenchmarkRuntimeConfig {
pub target_partitions: usize,
+ #[serde(default)]
+ pub parquet_pushdown_filters: bool,
pub memory_limit_bytes: Option<usize>,
pub spill_dir: Option<PathBuf>,
pub max_spill_bytes: Option<u64>,
@@ -39,6 +41,7 @@ impl Default for BenchmarkRuntimeConfig {
target_partitions: std::thread::available_parallelism()
.map(usize::from)
.unwrap_or(1),
+ parquet_pushdown_filters: false,
memory_limit_bytes: None,
spill_dir: None,
max_spill_bytes: None,
@@ -60,10 +63,15 @@ pub fn build_sql_context(config: &BenchmarkRuntimeConfig)
-> DataFusionResult<SQ
let sql = SQLContext::new();
let state_ref = sql.ctx().state_ref();
let current_state = state_ref.read().clone();
- let session_config = current_state
+ let mut session_config = current_state
.config()
.clone()
.with_target_partitions(config.target_partitions.max(1));
+ session_config
+ .options_mut()
+ .execution
+ .parquet
+ .pushdown_filters = config.parquet_pushdown_filters;
let state = SessionStateBuilder::from(current_state)
.with_config(session_config)
.with_runtime_env(Arc::new(runtime.build()?))
diff --git a/benchmarks/tpcds/src/lib.rs b/benchmarks/tpcds/src/lib.rs
index 6229b550..39442f86 100644
--- a/benchmarks/tpcds/src/lib.rs
+++ b/benchmarks/tpcds/src/lib.rs
@@ -183,6 +183,7 @@ mod tests {
let spill_dir = TempDir::new().unwrap();
let ctx = build_sql_context(&BenchmarkRuntimeConfig {
target_partitions: 3,
+ parquet_pushdown_filters: true,
memory_limit_bytes: Some(32 * 1024 * 1024),
spill_dir: Some(spill_dir.path().to_path_buf()),
max_spill_bytes: Some(64 * 1024 * 1024),
@@ -197,6 +198,14 @@ mod tests {
.target_partitions,
3
);
+ assert!(
+ ctx.ctx()
+ .state()
+ .config_options()
+ .execution
+ .parquet
+ .pushdown_filters
+ );
assert!(matches!(
ctx.ctx().runtime_env().memory_pool.memory_limit(),
MemoryLimit::Finite(size) if size == 32 * 1024 * 1024
diff --git a/benchmarks/tpcds/tests/smoke.rs b/benchmarks/tpcds/tests/smoke.rs
index c86d73ef..2974a3d8 100644
--- a/benchmarks/tpcds/tests/smoke.rs
+++ b/benchmarks/tpcds/tests/smoke.rs
@@ -225,6 +225,7 @@ async fn
command_orchestration_loads_and_writes_a_run_report() {
.unwrap();
let runtime = RuntimeArgs {
target_partitions: Some(2),
+ parquet_pushdown_filters: false,
memory_limit_gib: None,
spill_dir: None,
max_spill_gib: None,
diff --git a/crates/integrations/datafusion/src/config.rs
b/crates/integrations/datafusion/src/config.rs
new file mode 100644
index 00000000..7903a4fa
--- /dev/null
+++ b/crates/integrations/datafusion/src/config.rs
@@ -0,0 +1,61 @@
+// 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 datafusion::common::config::ConfigExtension;
+use datafusion::common::{config_namespace, extensions_options};
+
+pub const PAIMON_ROW_FILTER: &str = "paimon.read.row_filter";
+
+config_namespace! {
+ /// Paimon read options.
+ pub struct PaimonReadOptions {
+ /// Apply pushed predicates as row filters inside Paimon readers.
+ pub row_filter: bool, default = false
+ }
+}
+
+extensions_options! {
+ /// Paimon-specific DataFusion session options.
+ pub struct PaimonConfig {
+ /// Options that control Paimon reads.
+ pub read: PaimonReadOptions, default = PaimonReadOptions::default()
+ }
+}
+
+impl ConfigExtension for PaimonConfig {
+ const PREFIX: &'static str = "paimon";
+}
+
+#[cfg(test)]
+mod tests {
+ use datafusion::config::ConfigOptions;
+
+ use super::*;
+
+ #[test]
+ fn paimon_row_filter_defaults_to_false_and_can_be_set() {
+ let mut options = ConfigOptions::default();
+ options.extensions.insert(PaimonConfig::default());
+
+ let config = options.extensions.get::<PaimonConfig>().unwrap();
+ assert!(!config.read.row_filter);
+
+ options.set("paimon.read.row_filter", "true").unwrap();
+ let config = options.extensions.get::<PaimonConfig>().unwrap();
+ assert!(config.read.row_filter);
+ }
+}
diff --git a/crates/integrations/datafusion/src/filter_pushdown.rs
b/crates/integrations/datafusion/src/filter_pushdown.rs
index cef12aa6..33fd2aca 100644
--- a/crates/integrations/datafusion/src/filter_pushdown.rs
+++ b/crates/integrations/datafusion/src/filter_pushdown.rs
@@ -410,7 +410,7 @@ fn reverse_comparison_operator(op: Operator) ->
Option<Operator> {
}
}
-fn scalar_to_datum(scalar: &ScalarValue, data_type: &DataType) ->
Option<Datum> {
+pub(crate) fn scalar_to_datum(scalar: &ScalarValue, data_type: &DataType) ->
Option<Datum> {
match data_type {
DataType::Boolean(_) => match scalar {
ScalarValue::Boolean(Some(value)) => Some(Datum::Bool(*value)),
diff --git a/crates/integrations/datafusion/src/lib.rs
b/crates/integrations/datafusion/src/lib.rs
index 05e4037a..6f24ff92 100644
--- a/crates/integrations/datafusion/src/lib.rs
+++ b/crates/integrations/datafusion/src/lib.rs
@@ -40,6 +40,7 @@ mod blob_descriptor_functions;
mod blob_reader;
mod blob_view;
mod catalog;
+pub mod config;
mod delete;
mod error;
mod filter_pushdown;
diff --git a/crates/integrations/datafusion/src/physical_plan/scan.rs
b/crates/integrations/datafusion/src/physical_plan/scan.rs
index f374c034..2aa72536 100644
--- a/crates/integrations/datafusion/src/physical_plan/scan.rs
+++ b/crates/integrations/datafusion/src/physical_plan/scan.rs
@@ -15,27 +15,53 @@
// specific language governing permissions and limitations
// under the License.
+use std::collections::HashSet;
use std::sync::Arc;
+use std::time::Duration;
-use datafusion::arrow::compute::cast;
+use datafusion::arrow::array::BooleanArray;
+use datafusion::arrow::compute::{cast, filter_record_batch};
use datafusion::arrow::datatypes::{
DataType as ArrowDataType, SchemaRef as ArrowSchemaRef, TimeUnit,
};
use datafusion::arrow::record_batch::{RecordBatch, RecordBatchOptions};
use datafusion::common::stats::Precision;
use datafusion::common::{ColumnStatistics, ScalarValue, Statistics};
+use datafusion::config::ConfigOptions;
+use
datafusion::datasource::physical_plan::parquet::can_expr_be_pushed_down_with_schemas;
use datafusion::error::Result as DFResult;
use datafusion::execution::{SendableRecordBatchStream, TaskContext};
+use datafusion::logical_expr::Operator;
+use datafusion::physical_expr::expressions::{
+ BinaryExpr, Column, DynamicFilterPhysicalExpr, InListExpr, Literal,
+};
use datafusion::physical_expr::EquivalenceProperties;
+use datafusion::physical_expr::PhysicalExpr;
use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
+use datafusion::physical_plan::filter_pushdown::{
+ ChildPushdownResult, FilterPushdownPhase, FilterPushdownPropagation,
PushedDown,
+};
use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
use datafusion::physical_plan::{DisplayAs, ExecutionPlan, Partitioning,
PlanProperties};
+use futures::stream::FuturesUnordered;
use futures::{StreamExt, TryStreamExt};
-use paimon::spec::{DataField, Datum, MergeEngine, Predicate};
+use paimon::spec::{DataField, Datum, MergeEngine, Predicate, PredicateBuilder};
use paimon::table::{ScanTrace, Table};
use paimon::DataSplit;
+use crate::config::PaimonConfig;
use crate::error::to_datafusion_error;
+use crate::filter_pushdown::scalar_to_datum;
+
+const RUNTIME_FILTER_WAIT_MIN_ROWS: usize = 250_000;
+const RUNTIME_FILTER_WAIT_TIMEOUT: Duration = Duration::from_secs(1);
+
+fn scan_applies_row_filter(config: &ConfigOptions) -> bool {
+ config
+ .extensions
+ .get::<PaimonConfig>()
+ .is_some_and(|config| config.read.row_filter)
+}
fn to_datafusion_batch(batch: RecordBatch, schema: &ArrowSchemaRef) ->
DFResult<RecordBatch> {
if batch.num_columns() != schema.fields().len() {
@@ -64,6 +90,250 @@ fn to_datafusion_batch(batch: RecordBatch, schema:
&ArrowSchemaRef) -> DFResult<
RecordBatch::try_new_with_options(Arc::clone(schema), columns,
&options).map_err(Into::into)
}
+async fn runtime_pruning_predicate(
+ filters: &[Arc<dyn PhysicalExpr>],
+ fields: &[DataField],
+ case_sensitive: bool,
+ wait_timeout: Duration,
+) -> Option<Predicate> {
+ let mut pending = filters.to_vec();
+ let mut dynamic_filters = Vec::new();
+ while let Some(expr) = pending.pop() {
+ pending.extend(expr.children().into_iter().cloned());
+ if expr.downcast_ref::<DynamicFilterPhysicalExpr>().is_some() {
+ dynamic_filters.push(expr);
+ }
+ }
+
+ let mut waiters = FuturesUnordered::new();
+ for expr in dynamic_filters {
+ waiters.push(async move {
+ let dynamic = expr
+ .downcast_ref::<DynamicFilterPhysicalExpr>()
+ .expect("only dynamic filters are queued");
+ dynamic.wait_complete().await;
+ dynamic.expression_id()
+ });
+ }
+
+ let deadline = tokio::time::Instant::now() + wait_timeout;
+ let mut completed_dynamic_filters = HashSet::new();
+ while !waiters.is_empty() {
+ match tokio::time::timeout_at(deadline, waiters.next()).await {
+ Ok(Some(Some(expression_id))) => {
+ completed_dynamic_filters.insert(expression_id);
+ }
+ Ok(Some(None)) => {}
+ Ok(None) | Err(_) => break,
+ }
+ }
+
+ let predicate_builder = PredicateBuilder::new_with_case_sensitive(fields,
case_sensitive);
+ let mut predicates = Vec::new();
+ for filter in filters {
+ collect_runtime_pruning_predicates(
+ filter.as_ref(),
+ fields,
+ &predicate_builder,
+ case_sensitive,
+ &completed_dynamic_filters,
+ &mut predicates,
+ );
+ }
+ (!predicates.is_empty()).then(|| Predicate::and(predicates))
+}
+
+fn runtime_filter_wait_timeout(estimated_rows: usize) -> Duration {
+ if estimated_rows > RUNTIME_FILTER_WAIT_MIN_ROWS {
+ RUNTIME_FILTER_WAIT_TIMEOUT
+ } else {
+ Duration::ZERO
+ }
+}
+
+fn collect_runtime_pruning_predicates(
+ expr: &dyn PhysicalExpr,
+ fields: &[DataField],
+ predicate_builder: &PredicateBuilder,
+ case_sensitive: bool,
+ completed_dynamic_filters: &HashSet<u64>,
+ predicates: &mut Vec<Predicate>,
+) {
+ if let Some(dynamic) = expr.downcast_ref::<DynamicFilterPhysicalExpr>() {
+ if dynamic
+ .expression_id()
+ .is_none_or(|id| !completed_dynamic_filters.contains(&id))
+ {
+ return;
+ }
+ if let Ok(current) = dynamic.current() {
+ collect_runtime_pruning_predicates(
+ current.as_ref(),
+ fields,
+ predicate_builder,
+ case_sensitive,
+ completed_dynamic_filters,
+ predicates,
+ );
+ }
+ return;
+ }
+
+ if let Some(binary) = expr.downcast_ref::<BinaryExpr>() {
+ if binary.op() == &Operator::And {
+ collect_runtime_pruning_predicates(
+ binary.left().as_ref(),
+ fields,
+ predicate_builder,
+ case_sensitive,
+ completed_dynamic_filters,
+ predicates,
+ );
+ collect_runtime_pruning_predicates(
+ binary.right().as_ref(),
+ fields,
+ predicate_builder,
+ case_sensitive,
+ completed_dynamic_filters,
+ predicates,
+ );
+ } else if let Some(predicate) =
+ translate_runtime_comparison(binary, fields, predicate_builder,
case_sensitive)
+ {
+ predicates.push(predicate);
+ }
+ return;
+ }
+
+ if let Some(in_list) = expr.downcast_ref::<InListExpr>() {
+ if let Some(predicate) =
+ translate_runtime_in_list(in_list, fields, predicate_builder,
case_sensitive)
+ {
+ predicates.push(predicate);
+ }
+ }
+}
+
+fn translate_runtime_comparison(
+ binary: &BinaryExpr,
+ fields: &[DataField],
+ predicate_builder: &PredicateBuilder,
+ case_sensitive: bool,
+) -> Option<Predicate> {
+ let direct = runtime_column_literal(
+ binary.left().as_ref(),
+ binary.right().as_ref(),
+ fields,
+ case_sensitive,
+ )
+ .map(|(field, datum)| (*binary.op(), field, datum));
+ let comparison = direct.or_else(|| {
+ runtime_column_literal(
+ binary.right().as_ref(),
+ binary.left().as_ref(),
+ fields,
+ case_sensitive,
+ )
+ .and_then(|(field, datum)| {
+ reverse_runtime_comparison(*binary.op()).map(|op| (op, field,
datum))
+ })
+ })?;
+
+ let (op, field, datum) = comparison;
+ if matches!(
+ field.data_type(),
+ paimon::spec::DataType::Binary(_) |
paimon::spec::DataType::VarBinary(_)
+ ) && matches!(
+ op,
+ Operator::Lt | Operator::LtEq | Operator::Gt | Operator::GtEq
+ ) {
+ // Arrow compares binary values as unsigned bytes, while Paimon follows
+ // Java's signed-byte ordering. Range predicates could therefore prune
+ // rows that DataFusion would keep; equality predicates remain safe.
+ return None;
+ }
+ match op {
+ Operator::Eq => predicate_builder.equal(field.name(), datum).ok(),
+ Operator::NotEq => predicate_builder.not_equal(field.name(),
datum).ok(),
+ Operator::Lt => predicate_builder.less_than(field.name(), datum).ok(),
+ Operator::LtEq => predicate_builder.less_or_equal(field.name(),
datum).ok(),
+ Operator::Gt => predicate_builder.greater_than(field.name(),
datum).ok(),
+ Operator::GtEq => predicate_builder.greater_or_equal(field.name(),
datum).ok(),
+ _ => None,
+ }
+}
+
+fn translate_runtime_in_list(
+ in_list: &InListExpr,
+ fields: &[DataField],
+ predicate_builder: &PredicateBuilder,
+ case_sensitive: bool,
+) -> Option<Predicate> {
+ let column = in_list.expr().downcast_ref::<Column>()?;
+ let field = resolve_runtime_field(column.name(), fields, case_sensitive)?;
+ let literals = in_list
+ .list()
+ .iter()
+ .map(|expr| {
+ let literal = expr.downcast_ref::<Literal>()?;
+ if literal.value().is_null() {
+ return None;
+ }
+ scalar_to_datum(literal.value(), field.data_type())
+ })
+ .collect::<Option<Vec<_>>>()?;
+
+ if in_list.negated() {
+ predicate_builder.is_not_in(field.name(), literals).ok()
+ } else {
+ predicate_builder.is_in(field.name(), literals).ok()
+ }
+}
+
+fn runtime_column_literal<'a>(
+ column: &dyn PhysicalExpr,
+ literal: &dyn PhysicalExpr,
+ fields: &'a [DataField],
+ case_sensitive: bool,
+) -> Option<(&'a DataField, Datum)> {
+ let column = column.downcast_ref::<Column>()?;
+ let literal = literal.downcast_ref::<Literal>()?;
+ if literal.value().is_null() {
+ return None;
+ }
+ let field = resolve_runtime_field(column.name(), fields, case_sensitive)?;
+ let datum = scalar_to_datum(literal.value(), field.data_type())?;
+ Some((field, datum))
+}
+
+fn resolve_runtime_field<'a>(
+ name: &str,
+ fields: &'a [DataField],
+ case_sensitive: bool,
+) -> Option<&'a DataField> {
+ if case_sensitive {
+ fields.iter().find(|field| field.name() == name)
+ } else {
+ let mut matches = fields
+ .iter()
+ .filter(|field| field.name().eq_ignore_ascii_case(name));
+ let field = matches.next()?;
+ matches.next().is_none().then_some(field)
+ }
+}
+
+fn reverse_runtime_comparison(op: Operator) -> Option<Operator> {
+ match op {
+ Operator::Eq => Some(Operator::Eq),
+ Operator::NotEq => Some(Operator::NotEq),
+ Operator::Lt => Some(Operator::Gt),
+ Operator::LtEq => Some(Operator::GtEq),
+ Operator::Gt => Some(Operator::Lt),
+ Operator::GtEq => Some(Operator::LtEq),
+ _ => None,
+ }
+}
+
#[derive(Debug)]
struct ColumnStatsAccumulator {
min_value: Option<Datum>,
@@ -249,7 +519,7 @@ fn datum_to_scalar(value: Datum, data_type: &ArrowDataType)
-> Option<ScalarValu
/// Planning is performed eagerly in
[`super::super::table::PaimonTableProvider::scan`],
/// and the resulting splits are distributed across DataFusion execution
partitions
/// so that DataFusion can schedule them in parallel.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
pub struct PaimonTableScan {
table: Table,
/// Full Paimon read type for nested or connector-defined projections.
@@ -274,6 +544,14 @@ pub struct PaimonTableScan {
/// Column-name case sensitivity carried from planning to execution so the
/// read path resolves names the same way the scan was planned.
case_sensitive: bool,
+ /// Physical filters retained from DataFusion's runtime filter-pushdown
pass.
+ /// They are always available for conservative reader pruning and are
+ /// evaluated exactly only when Paimon row filtering is enabled.
+ runtime_filters: Vec<Arc<dyn PhysicalExpr>>,
+ /// Planning-time decision that this scan evaluates `runtime_filters`
+ /// exactly. Stored on the plan so execution cannot observe a different
+ /// session setting after the parent FilterExec has been removed.
+ apply_row_filter: bool,
}
impl PaimonTableScan {
@@ -307,6 +585,8 @@ impl PaimonTableScan {
scan_trace,
pushed_variants,
case_sensitive,
+ runtime_filters: Vec::new(),
+ apply_row_filter: false,
}
}
@@ -347,8 +627,9 @@ impl PaimonTableScan {
return Statistics::unknown_column(&self.schema());
}
- let exact_null_counts = (self.table.schema().primary_keys().is_empty()
- || merge_engine == MergeEngine::Deduplicate)
+ let exact_null_counts = self.runtime_filters.is_empty()
+ && (self.table.schema().primary_keys().is_empty()
+ || merge_engine == MergeEngine::Deduplicate)
&& self.pushed_predicate.is_none()
&& self.limit.is_none()
&& partitions
@@ -403,6 +684,59 @@ impl ExecutionPlan for PaimonTableScan {
Ok(self)
}
+ fn handle_child_pushdown_result(
+ &self,
+ _phase: FilterPushdownPhase,
+ child_pushdown_result: ChildPushdownResult,
+ config: &ConfigOptions,
+ ) -> DFResult<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
+ let filters = child_pushdown_result
+ .parent_filters
+ .into_iter()
+ .map(|result| result.filter)
+ .collect::<Vec<_>>();
+ if filters.is_empty() {
+ return Ok(FilterPushdownPropagation::with_parent_pushdown_result(
+ Vec::new(),
+ ));
+ }
+
+ let schema = self.schema();
+ let apply_row_filter = scan_applies_row_filter(config);
+ let mut accepted = Vec::new();
+ let parent_filter_handled = filters
+ .into_iter()
+ .map(|filter| {
+ if can_expr_be_pushed_down_with_schemas(&filter,
schema.as_ref()) {
+ accepted.push(filter);
+ // `PushedDown` reports whether this scan evaluates the
predicate
+ // exactly so the parent FilterExec can be removed. The
predicate is
+ // retained above for pruning regardless of this result.
+ if apply_row_filter {
+ PushedDown::Yes
+ } else {
+ PushedDown::No
+ }
+ } else {
+ PushedDown::No
+ }
+ })
+ .collect::<Vec<_>>();
+ if accepted.is_empty() {
+ return Ok(FilterPushdownPropagation::with_parent_pushdown_result(
+ parent_filter_handled,
+ ));
+ }
+
+ let mut scan = self.clone();
+ scan.runtime_filters.extend(accepted);
+ scan.apply_row_filter = apply_row_filter;
+ Ok(
+
FilterPushdownPropagation::with_parent_pushdown_result(parent_filter_handled)
+ .with_updated_node(Arc::new(scan)),
+ )
+ }
+
fn execute(
&self,
partition: usize,
@@ -420,23 +754,60 @@ impl ExecutionPlan for PaimonTableScan {
let read_type = self.read_type.clone();
let pushed_predicate = self.pushed_predicate.clone();
let case_sensitive = self.case_sensitive;
+ let runtime_filters = self.runtime_filters.clone();
+ let apply_row_filter = self.apply_row_filter;
let fut = async move {
let mut read_builder = table.new_read_builder();
read_builder.with_case_sensitive(case_sensitive);
read_builder.with_read_type(read_type);
- if let Some(filter) = pushed_predicate {
+ let estimated_rows = splits
+ .iter()
+ .filter_map(|split| usize::try_from(split.row_count()).ok())
+ .fold(0usize, usize::saturating_add);
+ let runtime_pruning = runtime_pruning_predicate(
+ &runtime_filters,
+ table.schema().fields(),
+ case_sensitive,
+ runtime_filter_wait_timeout(estimated_rows),
+ )
+ .await;
+ let read_predicate = match (pushed_predicate, runtime_pruning) {
+ (Some(pushed), Some(runtime)) =>
Some(Predicate::and(vec![pushed, runtime])),
+ (Some(predicate), None) | (None, Some(predicate)) =>
Some(predicate),
+ (None, None) => None,
+ };
+ if let Some(filter) = read_predicate {
read_builder.with_filter(filter);
}
- let read = read_builder.new_read().map_err(to_datafusion_error)?;
+ let read = read_builder
+ .new_read()
+ .map_err(to_datafusion_error)?
+ .with_row_filter(apply_row_filter);
let stream = read.to_arrow(&splits).map_err(to_datafusion_error)?;
let batch_schema = Arc::clone(&schema);
let stream = stream.map(move |result| {
- result
+ let mut batch = result
.map_err(to_datafusion_error)
- .and_then(|batch| to_datafusion_batch(batch,
&batch_schema))
+ .and_then(|batch| to_datafusion_batch(batch,
&batch_schema))?;
+ if apply_row_filter {
+ for filter in &runtime_filters {
+ let predicate =
filter.evaluate(&batch)?.into_array(batch.num_rows())?;
+ let predicate = predicate
+ .as_any()
+ .downcast_ref::<BooleanArray>()
+ .ok_or_else(|| {
+
datafusion::error::DataFusionError::Execution(format!(
+ "Paimon runtime filter must return
Boolean, got {}",
+ predicate.data_type()
+ ))
+ })?;
+ batch = filter_record_batch(&batch, predicate)?;
+ }
+ }
+ Ok(batch)
});
Ok::<_,
datafusion::error::DataFusionError>(RecordBatchStreamAdapter::new(
@@ -474,12 +845,15 @@ impl ExecutionPlan for PaimonTableScan {
// 1. All splits have known merged_row_count (no deletion files with
unknown cardinality)
// 2. No limit is applied (limit would make row count inexact)
// 3. Filter is exact (no residual filtering needed above the scan)
- let num_rows_precision =
- if all_row_counts_known && self.limit.is_none() &&
self.filter_exact {
- Precision::Exact(total_rows)
- } else {
- Precision::Inexact(total_rows)
- };
+ let num_rows_precision = if all_row_counts_known
+ && self.limit.is_none()
+ && self.filter_exact
+ && self.runtime_filters.is_empty()
+ {
+ Precision::Exact(total_rows)
+ } else {
+ Precision::Inexact(total_rows)
+ };
Ok(Arc::new(Statistics {
num_rows: num_rows_precision,
@@ -528,6 +902,14 @@ impl DisplayAs for PaimonTableScan {
if let Some(ref pushed_variants) = self.pushed_variants {
write!(f, ", PushedVariants=[{pushed_variants}]")?;
}
+ if !self.runtime_filters.is_empty() {
+ let filters = self
+ .runtime_filters
+ .iter()
+ .map(ToString::to_string)
+ .collect::<Vec<_>>();
+ write!(f, ", runtime_filters=[{}]", filters.join(" AND "))?;
+ }
Ok(())
}
}
@@ -535,20 +917,30 @@ impl DisplayAs for PaimonTableScan {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::config::PaimonConfig;
mod test_utils {
include!(concat!(env!("CARGO_MANIFEST_DIR"), "/../../test_utils.rs"));
}
use datafusion::arrow::array::Int32Array;
use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field,
Schema as ArrowSchema};
+ use datafusion::config::ConfigOptions;
+ use datafusion::logical_expr::Operator;
+ use datafusion::physical_expr::expressions::{
+ lit, BinaryExpr, Column, DynamicFilterPhysicalExpr, InListExpr,
+ };
+ use datafusion::physical_expr::PhysicalExpr;
+ use datafusion::physical_plan::filter_pushdown::{
+ ChildFilterPushdownResult, ChildPushdownResult,
+ };
use datafusion::physical_plan::ExecutionPlan;
use datafusion::prelude::SessionContext;
use futures::TryStreamExt;
use paimon::catalog::Identifier;
use paimon::io::FileIOBuilder;
use paimon::spec::{
- BinaryRow, DataFileMeta, DataType, Datum, IntType, PredicateBuilder,
- Schema as PaimonSchema, TableSchema,
+ BinaryRow, BinaryType, DataFileMeta, DataType, Datum, IntType,
PredicateBuilder,
+ Schema as PaimonSchema, TableSchema, VarBinaryType,
};
use paimon::table::{DeletionFile, RowRange, Table};
use std::fs;
@@ -585,6 +977,41 @@ mod tests {
}
}
+ #[test]
+ fn test_binary_runtime_ranges_are_not_translated() {
+ for data_type in [
+ DataType::Binary(BinaryType::new(1).unwrap()),
+ DataType::VarBinary(VarBinaryType::new(1).unwrap()),
+ ] {
+ let fields = vec![DataField::new(0, "bytes".to_string(),
data_type)];
+ let predicate_builder = PredicateBuilder::new(&fields);
+ for op in [Operator::Lt, Operator::LtEq, Operator::Gt,
Operator::GtEq] {
+ let expression = BinaryExpr::new(
+ Arc::new(Column::new("bytes", 0)),
+ op,
+
Arc::new(Literal::new(ScalarValue::Binary(Some(vec![0xff])))),
+ );
+
+ assert!(
+ translate_runtime_comparison(&expression, &fields,
&predicate_builder, true,)
+ .is_none(),
+ "binary {op} must remain with DataFusion"
+ );
+ }
+
+ let equality = BinaryExpr::new(
+ Arc::new(Column::new("bytes", 0)),
+ Operator::Eq,
+ Arc::new(Literal::new(ScalarValue::Binary(Some(vec![0xff])))),
+ );
+ assert!(
+ translate_runtime_comparison(&equality, &fields,
&predicate_builder, true)
+ .is_some(),
+ "binary equality is ordering-independent"
+ );
+ }
+ }
+
#[test]
fn test_partition_count_empty_plan() {
let schema = test_schema();
@@ -856,8 +1283,366 @@ mod tests {
assert_eq!(value_stats.null_count, Precision::Absent);
}
+ #[test]
+ fn test_scan_retains_dynamic_filter_for_runtime_evaluation() {
+ let scan = PaimonTableScan::new(
+ test_schema(),
+ dummy_table(),
+ test_read_type(),
+ None,
+ vec![Arc::from(Vec::<DataSplit>::new())],
+ None,
+ true,
+ None,
+ None,
+ true,
+ );
+ let dynamic_filter: Arc<dyn PhysicalExpr> =
+ Arc::new(DynamicFilterPhysicalExpr::new(Vec::new(), lit(true)));
+ let result = scan
+ .handle_child_pushdown_result(
+ FilterPushdownPhase::Post,
+ ChildPushdownResult {
+ parent_filters: vec![ChildFilterPushdownResult {
+ filter: dynamic_filter,
+ child_results: Vec::new(),
+ }],
+ self_filters: Vec::new(),
+ },
+ &ConfigOptions::default(),
+ )
+ .unwrap();
+
+ let updated = result
+ .updated_node
+ .expect("the scan must retain physical filters so dynamic join
filters remain active");
+ let statistics = updated.partition_statistics(None).unwrap();
+ assert_eq!(statistics.num_rows, Precision::Inexact(0));
+ assert_eq!(
+ statistics.column_statistics[0].null_count,
+ Precision::Absent,
+ "runtime-filtered scans must not expose unfiltered exact null
counts"
+ );
+ }
+
+ #[test]
+ fn test_scan_reports_filter_handled_when_row_filter_is_enabled() {
+ let scan = PaimonTableScan::new(
+ test_schema(),
+ dummy_table(),
+ test_read_type(),
+ None,
+ vec![Arc::from(Vec::<DataSplit>::new())],
+ None,
+ true,
+ None,
+ None,
+ true,
+ );
+ let filter: Arc<dyn PhysicalExpr> = Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("id", 0)),
+ Operator::Gt,
+ lit(1_i32),
+ ));
+ let mut paimon_config = PaimonConfig::default();
+ paimon_config.read.row_filter = true;
+ let mut config = ConfigOptions::default();
+ config.extensions.insert(paimon_config);
+
+ let result = scan
+ .handle_child_pushdown_result(
+ FilterPushdownPhase::Post,
+ ChildPushdownResult {
+ parent_filters: vec![ChildFilterPushdownResult {
+ filter,
+ child_results: Vec::new(),
+ }],
+ self_filters: Vec::new(),
+ },
+ &config,
+ )
+ .unwrap();
+
+ assert!(matches!(result.filters.as_slice(), [PushedDown::Yes]));
+ assert!(result.updated_node.is_some());
+ }
+
+ #[test]
+ fn test_scan_rejects_filter_outside_output_schema() {
+ let scan = PaimonTableScan::new(
+ test_schema(),
+ dummy_table(),
+ test_read_type(),
+ None,
+ vec![Arc::from(Vec::<DataSplit>::new())],
+ None,
+ true,
+ None,
+ None,
+ true,
+ );
+ let filter: Arc<dyn PhysicalExpr> = Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("missing", 1)),
+ Operator::Gt,
+ lit(1_i32),
+ ));
+ let mut paimon_config = PaimonConfig::default();
+ paimon_config.read.row_filter = true;
+ let mut config = ConfigOptions::default();
+ config.extensions.insert(paimon_config);
+
+ let result = scan
+ .handle_child_pushdown_result(
+ FilterPushdownPhase::Post,
+ ChildPushdownResult {
+ parent_filters: vec![ChildFilterPushdownResult {
+ filter,
+ child_results: Vec::new(),
+ }],
+ self_filters: Vec::new(),
+ },
+ &config,
+ )
+ .unwrap();
+
+ assert!(matches!(result.filters.as_slice(), [PushedDown::No]));
+ assert!(result.updated_node.is_none());
+ }
+
+ #[tokio::test]
+ async fn test_scan_uses_retained_runtime_filter_for_pruning_only() {
+ let tempdir = tempdir().unwrap();
+ let table_path = local_file_path(tempdir.path());
+ let bucket_dir = tempdir.path().join("bucket-0");
+ fs::create_dir_all(&bucket_dir).unwrap();
+ write_int_parquet_file(
+ &bucket_dir.join("data.parquet"),
+ vec![("id", vec![1, 2, 3, 4, 5])],
+ None,
+ );
+ let file_size =
fs::metadata(bucket_dir.join("data.parquet")).unwrap().len() as i64;
+ let file_io = FileIOBuilder::new("file").build().unwrap();
+ let table_schema = TableSchema::new(
+ 0,
+ &PaimonSchema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .build()
+ .unwrap(),
+ );
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "t"),
+ table_path,
+ table_schema,
+ None,
+ );
+ let split = paimon::DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(local_file_path(&bucket_dir))
+ .with_total_buckets(1)
+ .with_data_files(vec![test_data_file("data.parquet", 5,
file_size)])
+ .build()
+ .unwrap();
+ let scan = PaimonTableScan::new(
+ test_schema(),
+ table,
+ test_read_type(),
+ None,
+ vec![Arc::from(vec![split])],
+ None,
+ false,
+ None,
+ None,
+ true,
+ );
+ let filter: Arc<dyn PhysicalExpr> = Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("id", 0)),
+ Operator::Gt,
+ lit(2_i32),
+ ));
+ let pruning_result = scan
+ .handle_child_pushdown_result(
+ FilterPushdownPhase::Post,
+ ChildPushdownResult {
+ parent_filters: vec![ChildFilterPushdownResult {
+ filter: Arc::clone(&filter),
+ child_results: Vec::new(),
+ }],
+ self_filters: Vec::new(),
+ },
+ &ConfigOptions::default(),
+ )
+ .unwrap();
+ assert!(matches!(
+ pruning_result.filters.as_slice(),
+ [PushedDown::No]
+ ));
+ let pruning_scan = pruning_result.updated_node.unwrap();
+ let ctx = SessionContext::new();
+ let batches = pruning_scan
+ .execute(0, ctx.task_ctx())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let ids = collect_ids(&batches);
+
+ assert_eq!(
+ ids,
+ vec![1, 2, 3, 4, 5],
+ "unsupported physical filters may prune row groups but must not
remove rows"
+ );
+
+ let mut paimon_config = PaimonConfig::default();
+ paimon_config.read.row_filter = true;
+ let mut exact_config = ConfigOptions::default();
+ exact_config.extensions.insert(paimon_config);
+ let exact_result = scan
+ .handle_child_pushdown_result(
+ FilterPushdownPhase::Post,
+ ChildPushdownResult {
+ parent_filters: vec![ChildFilterPushdownResult {
+ filter,
+ child_results: Vec::new(),
+ }],
+ self_filters: Vec::new(),
+ },
+ &exact_config,
+ )
+ .unwrap();
+ assert!(matches!(exact_result.filters.as_slice(), [PushedDown::Yes]));
+
+ // Execution deliberately uses the default context. The scan must honor
+ // the planning-time decision that allowed the parent FilterExec to be
removed.
+ let exact_scan = exact_result.updated_node.unwrap();
+ let exact_batches = exact_scan
+ .execute(0, SessionContext::new().task_ctx())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(collect_ids(&exact_batches), vec![3, 4, 5]);
+ }
+
+ fn collect_ids(batches: &[RecordBatch]) -> Vec<i32> {
+ batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .iter()
+ .copied()
+ })
+ .collect()
+ }
+
+ #[tokio::test]
+ async fn test_dynamic_filter_builds_reader_pruning_predicate() {
+ let column: Arc<dyn PhysicalExpr> = Arc::new(Column::new("id", 0));
+ let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new(
+ vec![Arc::clone(&column)],
+ lit(true),
+ ));
+ dynamic_filter
+ .update(Arc::new(BinaryExpr::new(
+ column,
+ Operator::GtEq,
+ lit(3_i32),
+ )))
+ .unwrap();
+ dynamic_filter.mark_complete();
+ let filter: Arc<dyn PhysicalExpr> = dynamic_filter;
+
+ let predicate =
+ runtime_pruning_predicate(&[filter], &test_read_type(), true,
Duration::ZERO)
+ .await
+ .expect("a completed column/literal dynamic filter should
prune the reader");
+
+ assert_eq!(predicate.to_string(), "id >= 3");
+ }
+
+ #[tokio::test]
+ async fn test_runtime_in_list_builds_reader_pruning_predicate() {
+ let filter: Arc<dyn PhysicalExpr> = Arc::new(
+ InListExpr::try_new(
+ Arc::new(Column::new("id", 0)),
+ vec![lit(1_i32), lit(3_i32)],
+ false,
+ test_schema().as_ref(),
+ )
+ .unwrap(),
+ );
+
+ let predicate =
+ runtime_pruning_predicate(&[filter], &test_read_type(), true,
Duration::ZERO)
+ .await
+ .expect("a literal IN list should prune the reader");
+
+ assert_eq!(predicate.to_string(), "id IN (1, 3)");
+ }
+
+ #[test]
+ fn test_runtime_filter_wait_is_cost_aware() {
+ assert_eq!(runtime_filter_wait_timeout(204_000), Duration::ZERO);
+ assert_eq!(runtime_filter_wait_timeout(250_000), Duration::ZERO);
+ assert_eq!(runtime_filter_wait_timeout(250_001),
Duration::from_secs(1));
+ }
+
+ #[tokio::test]
+ async fn
test_incomplete_dynamic_filter_does_not_materially_delay_scan_startup() {
+ let column: Arc<dyn PhysicalExpr> = Arc::new(Column::new("id", 0));
+ let dynamic_filter: Arc<dyn PhysicalExpr> =
+ Arc::new(DynamicFilterPhysicalExpr::new(vec![column], lit(true)));
+
+ let predicate = tokio::time::timeout(
+ Duration::from_millis(100),
+ runtime_pruning_predicate(&[dynamic_filter], &test_read_type(),
true, Duration::ZERO),
+ )
+ .await
+ .expect("an incomplete dynamic filter must not delay scan startup");
+
+ assert!(predicate.is_none());
+ }
+
+ #[tokio::test]
+ async fn
test_completed_dynamic_filter_still_prunes_when_another_times_out() {
+ let column: Arc<dyn PhysicalExpr> = Arc::new(Column::new("id", 0));
+ let completed = Arc::new(DynamicFilterPhysicalExpr::new(
+ vec![Arc::clone(&column)],
+ lit(true),
+ ));
+ completed
+ .update(Arc::new(BinaryExpr::new(
+ Arc::clone(&column),
+ Operator::GtEq,
+ lit(3_i32),
+ )))
+ .unwrap();
+ completed.mark_complete();
+ let incomplete: Arc<dyn PhysicalExpr> =
+ Arc::new(DynamicFilterPhysicalExpr::new(vec![column], lit(true)));
+ let completed: Arc<dyn PhysicalExpr> = completed;
+
+ let predicate = runtime_pruning_predicate(
+ &[completed, incomplete],
+ &test_read_type(),
+ true,
+ Duration::ZERO,
+ )
+ .await
+ .expect("completed filters should survive another filter timing out");
+
+ assert_eq!(predicate.to_string(), "id >= 3");
+ }
+
#[tokio::test]
- async fn test_execute_applies_pushed_filter_during_read() {
+ async fn test_execute_uses_pushed_filter_only_for_pruning_by_default() {
let tempdir = tempdir().unwrap();
let table_path = local_file_path(tempdir.path());
let bucket_dir = tempdir.path().join("bucket-0");
@@ -941,7 +1726,7 @@ mod tests {
})
.collect();
- assert_eq!(actual_ids, vec![2, 3, 4]);
+ assert_eq!(actual_ids, vec![1, 2, 3, 4]);
}
#[tokio::test]
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index 3190487b..910edc8a 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -106,7 +106,10 @@ impl SQLContext {
/// Creates a new empty SQL context.
pub fn new() -> Self {
let state = SessionStateBuilder::new()
- .with_config(crate::lateral_vector_search::session_config())
+ .with_config(
+ crate::lateral_vector_search::session_config()
+
.with_option_extension(crate::config::PaimonConfig::default()),
+ )
.with_default_features()
.with_relation_planners(vec![Arc::new(
crate::relation_planner::PaimonRelationPlanner::new(),
@@ -412,6 +415,9 @@ impl SQLContext {
}) => {
let key = variable.to_string();
let key = key.trim_matches('\'').trim_matches('"');
+ if key == crate::config::PAIMON_ROW_FILTER {
+ return self.ctx.sql(sql).await;
+ }
if let Some(paimon_key) = key.strip_prefix("paimon.") {
let value = values
.first()
@@ -435,6 +441,9 @@ impl SQLContext {
}) => {
let key = name.to_string();
let key = key.trim_matches('\'').trim_matches('"');
+ if key == crate::config::PAIMON_ROW_FILTER {
+ return self.ctx.sql("SET paimon.read.row_filter =
false").await;
+ }
if let Some(paimon_key) = key.strip_prefix("paimon.") {
self.dynamic_options.write().unwrap().remove(paimon_key);
return ok_result(&self.ctx);
@@ -6572,6 +6581,50 @@ mod tests {
assert!(opts.is_empty());
}
+ #[tokio::test]
+ async fn test_set_paimon_read_option_delegates_to_session_config() {
+ let catalog = Arc::new(MockCatalog::new());
+ let sql_context = make_sql_context(catalog).await;
+
+ sql_context
+ .sql("SET paimon.read.row_filter = true")
+ .await
+ .unwrap();
+
+ let state = sql_context.ctx().state();
+ let config = state
+ .config_options()
+ .extensions
+ .get::<crate::config::PaimonConfig>()
+ .unwrap();
+ assert!(config.read.row_filter);
+ assert!(sql_context.dynamic_options().read().unwrap().is_empty());
+ }
+
+ #[tokio::test]
+ async fn test_reset_paimon_read_option_delegates_to_session_config() {
+ let catalog = Arc::new(MockCatalog::new());
+ let sql_context = make_sql_context(catalog).await;
+
+ sql_context
+ .sql("SET paimon.read.row_filter = true")
+ .await
+ .unwrap();
+ sql_context
+ .sql("RESET paimon.read.row_filter")
+ .await
+ .unwrap();
+
+ let state = sql_context.ctx().state();
+ let config = state
+ .config_options()
+ .extensions
+ .get::<crate::config::PaimonConfig>()
+ .unwrap();
+ assert!(!config.read.row_filter);
+ assert!(sql_context.dynamic_options().read().unwrap().is_empty());
+ }
+
#[tokio::test]
async fn test_set_multiple_paimon_options() {
let catalog = Arc::new(MockCatalog::new());
diff --git a/crates/integrations/datafusion/tests/blob_tests.rs
b/crates/integrations/datafusion/tests/blob_tests.rs
index 59058d9b..5df82d69 100644
--- a/crates/integrations/datafusion/tests/blob_tests.rs
+++ b/crates/integrations/datafusion/tests/blob_tests.rs
@@ -883,7 +883,7 @@ async fn
test_blob_descriptor_field_short_read_returns_error() {
}
#[tokio::test]
-async fn
test_blob_descriptor_filter_before_resolve_skips_filtered_bad_descriptor() {
+async fn test_blob_descriptor_filter_before_resolve_when_pushdown_enabled() {
let (tmp, sql_context) = setup(
"CREATE TABLE paimon.test_db.t (\
id INT, \
@@ -906,6 +906,7 @@ async fn
test_blob_descriptor_filter_before_resolve_skips_filtered_bad_descripto
(2, 'Kept', X'4F4B')"
);
exec(&sql_context, &sql).await;
+ exec(&sql_context, "SET paimon.read.row_filter = true").await;
let rows = query_id_name_picture(
&sql_context,
diff --git a/crates/paimon/src/arrow/format/avro.rs
b/crates/paimon/src/arrow/format/avro.rs
index 73dda9c6..d58f13b3 100644
--- a/crates/paimon/src/arrow/format/avro.rs
+++ b/crates/paimon/src/arrow/format/avro.rs
@@ -64,6 +64,7 @@ impl FormatFileReader for AvroFormatReader {
// caller's `&FilePredicates` (FilePredicates is not `Clone`; rebuild
it).
let predicates = predicates.map(|fp| FilePredicates {
predicates: fp.predicates.clone(),
+ apply_row_filter: fp.apply_row_filter,
file_fields: fp.file_fields.clone(),
});
@@ -1029,6 +1030,7 @@ mod tests {
PredicateOperator::Gt,
vec![Datum::Long(25)],
)],
+ apply_row_filter: true,
file_fields,
};
@@ -1089,6 +1091,7 @@ mod tests {
PredicateOperator::Gt,
vec![Datum::Long(25)],
)],
+ apply_row_filter: true,
file_fields,
};
@@ -1157,6 +1160,7 @@ mod tests {
PredicateOperator::Like,
vec![Datum::String("a%".to_string())],
)],
+ apply_row_filter: true,
file_fields: vec![age, name],
};
diff --git a/crates/paimon/src/arrow/format/mod.rs
b/crates/paimon/src/arrow/format/mod.rs
index ad63e0cc..b35f96f7 100644
--- a/crates/paimon/src/arrow/format/mod.rs
+++ b/crates/paimon/src/arrow/format/mod.rs
@@ -43,6 +43,8 @@ use std::collections::HashMap;
pub(crate) struct FilePredicates {
/// Predicates with indices already remapped to file-level fields.
pub predicates: Vec<Predicate>,
+ /// Whether predicates may remove individual rows from emitted batches.
+ pub apply_row_filter: bool,
/// File-level fields (full file schema), used for stats access and row
filtering.
pub file_fields: Vec<DataField>,
}
@@ -62,13 +64,17 @@ pub(crate) trait FormatFileReader: Send + Sync {
/// for residual filtering); the caller (`DataFileReader`) projects to the
/// requested output by name, so extra columns are harmless.
///
- /// Predicate exactness is per-format, NOT a blanket guarantee:
+ /// When `FilePredicates::apply_row_filter` is true, predicate exactness is
+ /// per-format, NOT a blanket guarantee:
/// - Parquet, ORC, Avro, Row, 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.
+ /// - When `apply_row_filter` is false, formats may still use predicates
for
+ /// conservative file, row-group, stripe, or page pruning, but must not
+ /// remove individual rows from a retained unit.
///
/// `row_selection` is a pre-merged list of 0-based inclusive row ranges
/// (DV + row_ranges already combined by the caller).
diff --git a/crates/paimon/src/arrow/format/mosaic.rs
b/crates/paimon/src/arrow/format/mosaic.rs
index 0242475d..fee2d22b 100644
--- a/crates/paimon/src/arrow/format/mosaic.rs
+++ b/crates/paimon/src/arrow/format/mosaic.rs
@@ -60,6 +60,7 @@ impl FormatFileReader for MosaicFormatReader {
let read_fields = read_fields.to_vec();
let predicates = predicates.map(|predicates| FilePredicates {
predicates: predicates.predicates.clone(),
+ apply_row_filter: predicates.apply_row_filter,
file_fields: predicates.file_fields.clone(),
});
let batch_size = batch_size.unwrap_or(DEFAULT_BATCH_SIZE);
@@ -867,6 +868,7 @@ mod tests {
) -> FilePredicates {
FilePredicates {
predicates,
+ apply_row_filter: true,
file_fields: fields,
}
}
diff --git a/crates/paimon/src/arrow/format/orc.rs
b/crates/paimon/src/arrow/format/orc.rs
index 81556b14..69f3edb7 100644
--- a/crates/paimon/src/arrow/format/orc.rs
+++ b/crates/paimon/src/arrow/format/orc.rs
@@ -97,6 +97,7 @@ impl FormatFileReader for OrcFormatReader {
(
FilePredicates {
predicates: fp.predicates.clone(),
+ apply_row_filter: fp.apply_row_filter,
file_fields: fp.file_fields.clone(),
},
scan_fields,
@@ -424,6 +425,7 @@ mod tests {
fn file_predicates(predicates: Vec<Predicate>, file_fields:
Vec<DataField>) -> FilePredicates {
FilePredicates {
predicates,
+ apply_row_filter: true,
file_fields,
}
}
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index c15ead0d..bb30b879 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -270,10 +270,12 @@ impl FormatFileReader for ParquetFormatReader {
let arrow_file_reader = ArrowFileReader::new(file_size, reader);
let empty_predicates = Vec::new();
- let (preds, file_fields): (&[Predicate], &[DataField]) = match
predicates {
- Some(fp) => (&fp.predicates, &fp.file_fields),
- None => (&empty_predicates, &[]),
- };
+ let (preds, apply_row_filter, file_fields): (&[Predicate], bool,
&[DataField]) =
+ match predicates {
+ Some(fp) => (&fp.predicates, fp.apply_row_filter,
&fp.file_fields),
+ None => (&empty_predicates, true, &[]),
+ };
+ let pruning_preds = preds;
// Only load the Parquet page index (ColumnIndex + OffsetIndex) when a
// predicate can use it for page-level pruning — matching Java Paimon,
@@ -284,7 +286,7 @@ impl FormatFileReader for ParquetFormatReader {
// skip it. `Optional` lets files without a page index fall through to
// row-group-level pruning instead of erroring.
let mut arrow_options = ArrowReaderOptions::new();
- if !preds.is_empty() {
+ if !pruning_preds.is_empty() {
arrow_options =
arrow_options.with_page_index_policy(PageIndexPolicy::Optional);
}
let mut batch_stream_builder =
@@ -303,9 +305,10 @@ impl FormatFileReader for ParquetFormatReader {
// predicate is fully enforced, so we decode exactly `read_fields`,
skip
// the residual pass entirely, and return the stream as before — zero
// added overhead.
- let all_enforced = preds
- .iter()
- .all(|p| predicate_fully_enforced_by_row_filter(&parquet_schema,
p, file_fields));
+ let all_enforced = !apply_row_filter
+ || preds
+ .iter()
+ .all(|p|
predicate_fully_enforced_by_row_filter(&parquet_schema, p, file_fields));
// Residual branch must decode the predicate columns too, or the
residual
// pass could not see a predicate on a non-projected column (Gap A).
The
@@ -331,14 +334,16 @@ impl FormatFileReader for ParquetFormatReader {
let mask = ProjectionMask::roots(&parquet_schema, root_indices);
batch_stream_builder = batch_stream_builder.with_projection(mask);
- let parquet_row_filter = build_parquet_row_filter(&parquet_schema,
preds, file_fields)?;
- if let Some(f) = parquet_row_filter {
- batch_stream_builder = batch_stream_builder.with_row_filter(f);
+ if apply_row_filter {
+ let parquet_row_filter = build_parquet_row_filter(&parquet_schema,
preds, file_fields)?;
+ if let Some(f) = parquet_row_filter {
+ batch_stream_builder = batch_stream_builder.with_row_filter(f);
+ }
}
let predicate_row_selection = build_predicate_row_selection(
batch_stream_builder.metadata().row_groups(),
- preds,
+ pruning_preds,
file_fields,
)?;
let mut combined_selection = predicate_row_selection;
@@ -346,8 +351,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 = build_predicate_page_selection(
+ batch_stream_builder.metadata(),
+ pruning_preds,
+ file_fields,
+ )?;
combined_selection =
intersect_optional_row_selections(combined_selection, page_selection);
if let Some(ref ranges) = row_selection {
@@ -396,6 +404,7 @@ impl FormatFileReader for ParquetFormatReader {
// projects the filtered batch to `read_fields` by name.
let residual_predicates = FilePredicates {
predicates: preds.to_vec(),
+ apply_row_filter: true,
file_fields: file_fields.to_vec(),
};
let stream = batch_stream.map(move |result| {
@@ -3009,6 +3018,7 @@ mod tests {
let predicates = FilePredicates {
predicates: vec![predicate],
+ apply_row_filter: true,
file_fields: id_name_age_file_fields(),
};
@@ -3163,6 +3173,7 @@ mod tests {
let reader_input = input.reader().await.unwrap();
let predicates = FilePredicates {
predicates: vec![leaf_gt, leaf_lt],
+ apply_row_filter: true,
file_fields,
};
let reader = ParquetFormatReader;
@@ -3480,6 +3491,7 @@ mod tests {
])];
let file_predicates = FilePredicates {
predicates,
+ apply_row_filter: true,
file_fields: fields.clone(),
};
diff --git a/crates/paimon/src/arrow/format/row.rs
b/crates/paimon/src/arrow/format/row.rs
index 94b740f4..a3fbd9d9 100644
--- a/crates/paimon/src/arrow/format/row.rs
+++ b/crates/paimon/src/arrow/format/row.rs
@@ -300,6 +300,7 @@ impl FormatFileReader for RowFormatReader {
let blocks_to_read = blocks_to_read(&index, total_rows,
row_selection.as_deref());
let predicates = predicates.map(|fp| FilePredicates {
predicates: fp.predicates.clone(),
+ apply_row_filter: fp.apply_row_filter,
file_fields: fp.file_fields.clone(),
});
Ok(try_stream! {
@@ -2417,6 +2418,7 @@ mod tests {
op: PredicateOperator::Gt,
literals: vec![Datum::Int(25)],
}],
+ apply_row_filter: true,
file_fields: fields.clone(),
};
@@ -2493,6 +2495,7 @@ mod tests {
op: PredicateOperator::Gt,
literals: vec![Datum::Int(25)],
}],
+ apply_row_filter: true,
file_fields: fields.clone(),
};
diff --git a/crates/paimon/src/arrow/format/vortex.rs
b/crates/paimon/src/arrow/format/vortex.rs
index 631f8add..e0d15cb1 100644
--- a/crates/paimon/src/arrow/format/vortex.rs
+++ b/crates/paimon/src/arrow/format/vortex.rs
@@ -76,6 +76,7 @@ impl FormatFileReader for VortexFormatReader {
let read_fields = read_fields.to_vec();
let predicates = predicates.map(|fp| FilePredicates {
predicates: fp.predicates.clone(),
+ apply_row_filter: fp.apply_row_filter,
file_fields: fp.file_fields.clone(),
});
let scan_fields = widen_scan_fields(&read_fields, predicates.as_ref());
@@ -1139,6 +1140,7 @@ mod tests {
let pred = builder.equal("id", Datum::Int(3)).unwrap();
let fp = FilePredicates {
predicates: vec![pred],
+ apply_row_filter: true,
file_fields: fields,
};
let ids =
@@ -1153,6 +1155,7 @@ mod tests {
let pred = builder.greater_than("id", Datum::Int(3)).unwrap();
let fp = FilePredicates {
predicates: vec![pred],
+ apply_row_filter: true,
file_fields: fields,
};
let ids =
@@ -1169,6 +1172,7 @@ mod tests {
.unwrap();
let fp = FilePredicates {
predicates: vec![pred],
+ apply_row_filter: true,
file_fields: fields,
};
let ids =
@@ -1185,6 +1189,7 @@ mod tests {
let pred2 = builder.less_than("value", Datum::Int(50)).unwrap();
let fp = FilePredicates {
predicates: vec![pred1, pred2],
+ apply_row_filter: true,
file_fields: fields,
};
let ids =
@@ -1201,6 +1206,7 @@ mod tests {
let pred = builder.equal("id", Datum::Int(99)).unwrap();
let fp = FilePredicates {
predicates: vec![pred],
+ apply_row_filter: true,
file_fields: fields,
};
let ids =
@@ -1236,6 +1242,7 @@ mod tests {
let pred = builder.greater_than("value", Datum::Int(30)).unwrap();
let fp = FilePredicates {
predicates: vec![pred],
+ apply_row_filter: true,
file_fields: fields.clone(),
};
let read_fields = vec![fields[0].clone()];
@@ -1277,6 +1284,7 @@ mod tests {
let pred = builder.greater_than("id", Datum::Int(3)).unwrap();
let fp = FilePredicates {
predicates: vec![pred],
+ apply_row_filter: true,
file_fields: fields,
};
@@ -1294,10 +1302,12 @@ mod tests {
let builder = PredicateBuilder::new(&fields);
let eq = FilePredicates {
predicates: vec![builder.equal("id", Datum::Int(3)).unwrap()],
+ apply_row_filter: true,
file_fields: fields.clone(),
};
let gt = FilePredicates {
predicates: vec![builder.greater_than("id",
Datum::Int(3)).unwrap()],
+ apply_row_filter: true,
file_fields: fields.clone(),
};
let combined = FilePredicates {
@@ -1305,6 +1315,7 @@ mod tests {
builder.greater_or_equal("id", Datum::Int(2)).unwrap(),
builder.less_than("value", Datum::Int(50)).unwrap(),
],
+ apply_row_filter: true,
file_fields: fields,
};
diff --git a/crates/paimon/src/arrow/residual.rs
b/crates/paimon/src/arrow/residual.rs
index 1a6ff5a1..b2259963 100644
--- a/crates/paimon/src/arrow/residual.rs
+++ b/crates/paimon/src/arrow/residual.rs
@@ -76,6 +76,10 @@ pub(crate) fn filter_record_batch_by_predicates(
predicates: &FilePredicates,
scan_fields: &[DataField],
) -> crate::Result<RecordBatch> {
+ if !predicates.apply_row_filter {
+ return Ok(batch);
+ }
+
let Some(mask) = evaluate_predicates_mask(
&batch,
&predicates.predicates,
@@ -250,7 +254,7 @@ pub(crate) fn widen_scan_fields(
) -> Vec<DataField> {
let mut fields = read_fields.to_vec();
- if let Some(fp) = predicates {
+ if let Some(fp) = predicates.filter(|fp| fp.apply_row_filter) {
let mut predicate_indices = Vec::new();
for predicate in &fp.predicates {
collect_predicate_field_indices(predicate, &mut predicate_indices);
@@ -943,6 +947,7 @@ mod tests {
fn file_predicates(predicates: Vec<Predicate>, file_fields:
Vec<DataField>) -> FilePredicates {
FilePredicates {
predicates,
+ apply_row_filter: true,
file_fields,
}
}
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 14871adf..ae99a5fc 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -103,9 +103,11 @@ pub(crate) struct DataEvolutionReader {
/// Arrow schema of wide batches at the _ROW_ID attach point: the original
/// read_type columns in caller order, then the extra predicate columns.
wide_output_schema: Arc<arrow_schema::Schema>,
- /// Data predicates (table-schema leaf indices). Applied exactly to every
- /// batch after _ROW_ID attachment, before yielding.
+ /// Data predicates (table-schema leaf indices). Available for pruning and,
+ /// when `apply_row_filter` is enabled, applied exactly after `_ROW_ID`
+ /// attachment before yielding.
predicates: Vec<Predicate>,
+ apply_row_filter: bool,
blob_as_descriptor: bool,
blob_descriptor_fields: HashSet<String>,
blob_view_fields: HashSet<String>,
@@ -143,8 +145,10 @@ impl DataEvolutionReader {
// at the END: `project_output` relies on that to project the
// final batch back to `read_type` by prefix. Predicate leaf indices
// point into the table schema, so `file_fields` = `table_fields`.
+ let apply_row_filter = true;
let file_predicates = (!predicates.is_empty()).then(|| FilePredicates {
predicates: predicates.clone(),
+ apply_row_filter,
file_fields: table_fields.clone(),
});
let wide_file_read_type =
@@ -167,6 +171,7 @@ impl DataEvolutionReader {
output_schema,
wide_output_schema,
predicates,
+ apply_row_filter,
blob_as_descriptor,
blob_descriptor_fields,
blob_view_fields,
@@ -177,6 +182,11 @@ impl DataEvolutionReader {
})
}
+ pub(crate) fn with_row_filter(mut self, apply_row_filter: bool) -> Self {
+ self.apply_row_filter = apply_row_filter;
+ self
+ }
+
pub(crate) fn with_batch_size(mut self, batch_size: Option<usize>) -> Self
{
self.batch_size = batch_size;
self
@@ -198,14 +208,26 @@ impl DataEvolutionReader {
let filter_before_blob_resolution =
self.can_filter_before_blob_resolution(blob_view_lookup.is_some(),
&descriptor_fields);
+ // Raw-convertible files can safely use predicates for conservative
+ // row-group/page pruning. Keep row filtering disabled here because
+ // the exact residual runs after schema evolution and `_ROW_ID`
+ // attachment. Positional row IDs cannot currently be reconciled
+ // with pruned row groups, so that projection falls back to no
+ // predicate pushdown.
+ let pruning_predicates = if self.row_id_index.is_none() {
+ self.predicates.clone()
+ } else {
+ Vec::new()
+ };
let file_reader = DataFileReader::new(
self.file_io.clone(),
self.schema_manager.clone(),
self.table_schema_id,
self.table_fields.clone(),
self.wide_file_read_type.clone(),
- Vec::new(),
+ pruning_predicates,
)
+ .with_row_filter(false)
.with_batch_size(self.batch_size);
for split in splits {
@@ -360,7 +382,7 @@ impl DataEvolutionReader {
/// `_ROW_ID` correctness: ids are attached before this filter, so
surviving
/// rows keep their original ids.
fn filter_wide_batch(&self, batch: RecordBatch) ->
crate::Result<RecordBatch> {
- if self.predicates.is_empty() {
+ if !self.apply_row_filter || self.predicates.is_empty() {
return Ok(batch);
}
@@ -448,6 +470,9 @@ impl DataEvolutionReader {
resolve_blob_views: bool,
descriptor_fields: &HashSet<String>,
) -> bool {
+ if !self.apply_row_filter {
+ return false;
+ }
let mut transformed_fields = HashSet::new();
if resolve_blob_views {
transformed_fields.extend(self.blob_view_fields.iter().cloned());
@@ -504,6 +529,7 @@ impl DataEvolutionReader {
false,
None,
)?
+ .with_row_filter(self.apply_row_filter)
.with_batch_size(self.batch_size);
let mut stream = prescan.read(splits)?;
let mut view_structs = HashSet::new();
@@ -5084,6 +5110,55 @@ mod tests {
assert_eq!(collect_int_values(&batches, "value"), vec![20, 30, 40]);
}
+ #[tokio::test]
+ async fn test_evolution_read_row_filter_disabled_still_prunes_row_groups()
{
+ let tempdir = tempdir().unwrap();
+ let table_path = local_file_path(tempdir.path());
+ let bucket_dir = tempdir.path().join("bucket-0");
+ fs::create_dir_all(&bucket_dir).unwrap();
+
+ let parquet_path = bucket_dir.join("data.parquet");
+ write_int_parquet_file(
+ &parquet_path,
+ vec![("id", vec![1, 2, 3, 4]), ("value", vec![5, 10, 30, 40])],
+ Some(2),
+ );
+
+ let table = two_col_evolution_table(table_path);
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(local_file_path(&bucket_dir))
+ .with_total_buckets(1)
+ .with_data_files(vec![data_file_meta_with_path(
+ "data.parquet",
+ 0,
+ 4,
+ 1,
+ parquet_path.metadata().unwrap().len() as i64,
+ Some(vec!["id", "value"]),
+ )])
+ .build()
+ .unwrap();
+
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .greater_than("id", Datum::Int(3))
+ .unwrap();
+ let mut builder = table.new_read_builder();
+ builder.with_filter(predicate);
+ let read = builder.new_read().unwrap().with_row_filter(false);
+ let batches = read
+ .to_arrow(&[split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(collect_int_values(&batches, "id"), vec![3, 4]);
+ assert_eq!(collect_int_values(&batches, "value"), vec![30, 40]);
+ }
+
/// Raw-convertible branch: a compound `Or` predicate referencing a
/// NON-projected column filters exactly, and the widened predicate column
/// does not leak into the output schema.
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 0faff2e9..369833b0 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -43,6 +43,7 @@ pub(crate) struct DataFileReader {
table_fields: Vec<DataField>,
read_type: Vec<DataField>,
predicates: Vec<Predicate>,
+ row_filter: bool,
blob_as_descriptor: bool,
batch_size: Option<usize>,
}
@@ -63,6 +64,7 @@ impl DataFileReader {
table_fields,
read_type,
predicates,
+ row_filter: true,
blob_as_descriptor: false,
batch_size: None,
}
@@ -78,6 +80,11 @@ impl DataFileReader {
self
}
+ pub(crate) fn with_row_filter(mut self, row_filter: bool) -> Self {
+ self.row_filter = row_filter;
+ self
+ }
+
/// Return a copy with a replaced read-type. Used by
`pk_vector_position_read`
/// to inject the internal `_ROW_ID` column for physical-position recovery.
pub(super) fn with_read_type(mut self, read_type: Vec<DataField>) -> Self {
@@ -94,7 +101,7 @@ impl DataFileReader {
/// True if any configured predicate can actually drop rows. A lone
/// `Predicate::AlwaysTrue` keeps every row in order and is not
row-filtering,
- /// matching `reject_row_id_with_predicate`'s notion. Consumed by
+ /// matching `reject_row_id_with_predicates`'s notion. Consumed by
/// `pk_vector_position_read` (materialization read path).
pub(super) fn has_row_filtering_predicate(&self) -> bool {
self.predicates
@@ -102,11 +109,11 @@ impl DataFileReader {
.any(|p| !matches!(p, Predicate::AlwaysTrue))
}
- /// Reject projecting `_ROW_ID` alongside a data predicate. `_ROW_ID` is
- /// assigned positionally from post-filter batch row counts, so a residual
- /// filter that drops rows would desync it. (`_ROW_ID` predicates travel
via
- /// `row_ranges`, not `predicates`, so they do not trip this.)
- fn reject_row_id_with_predicate(
+ /// Reject projecting `_ROW_ID` alongside an exact or pruning predicate.
+ /// `_ROW_ID` is assigned positionally from emitted batch row counts, so
+ /// residual filtering or row-group/page pruning would desync it.
(`_ROW_ID`
+ /// predicates travel via `row_ranges`, so they do not trip this.)
+ fn reject_row_id_with_predicates(
read_type: &[DataField],
predicates: &[Predicate],
) -> crate::Result<()> {
@@ -121,8 +128,9 @@ impl DataFileReader {
.any(|p| !matches!(p, Predicate::AlwaysTrue));
if projects_row_id && has_row_filtering_predicate {
return Err(crate::Error::Unsupported {
- message: "reading _ROW_ID together with a data predicate is
not supported yet"
- .to_string(),
+ message:
+ "reading _ROW_ID together with a data or pruning predicate
is not supported yet"
+ .to_string(),
});
}
Ok(())
@@ -238,19 +246,20 @@ impl DataFileReader {
) -> crate::Result<ArrowRecordBatchStream> {
// Guard at the true risk site: `_ROW_ID` is materialized positionally
from
// each batch's row count (see `row_id_column_for_batch`), assuming the
- // reader emits rows in original file order and count. The format
readers
- // apply an exact residual filter that drops non-matching rows *before*
- // `_ROW_ID` is assigned here, which would desync the ids. So
projecting
- // `_ROW_ID` together with a data predicate is unsupported — fail
loudly
+ // reader emits rows in original file order and count. Format readers
may
+ // skip row groups/pages or apply an exact row filter *before*
`_ROW_ID`
+ // is assigned here, which would desync the ids. So projecting
`_ROW_ID`
+ // together with a data predicate is unsupported — fail loudly
// rather than return wrong ids. Placed here (not only in `read()`)
because
// `read_single_file_stream` is also called directly by the KV and
// data-evolution readers; both strip/omit `_ROW_ID` from the read_type
// they pass, so this guard does not affect them.
- Self::reject_row_id_with_predicate(&self.read_type, &self.predicates)?;
+ Self::reject_row_id_with_predicates(&self.read_type,
&self.predicates)?;
let read_type = self.read_type.clone();
let table_fields = self.table_fields.clone();
let predicates = self.predicates.clone();
+ let row_filter = self.row_filter;
let file_io = self.file_io.clone();
let split = split.clone();
let blob_as_descriptor = self.blob_as_descriptor;
@@ -293,6 +302,7 @@ impl DataFileReader {
} else {
Some(crate::arrow::format::FilePredicates {
predicates: remapped,
+ apply_row_filter: row_filter,
file_fields: file_fields.clone(),
})
}
@@ -457,6 +467,7 @@ impl DataFileReader {
let read_type = self.read_type.clone();
let table_fields = self.table_fields.clone();
let predicates = self.predicates.clone();
+ let row_filter = self.row_filter;
let file_io = self.file_io.clone();
let split = split.clone();
let blob_as_descriptor = self.blob_as_descriptor;
@@ -498,6 +509,7 @@ impl DataFileReader {
} else {
Some(crate::arrow::format::FilePredicates {
predicates: remapped,
+ apply_row_filter: row_filter,
file_fields: file_fields.clone(),
})
}
@@ -1163,6 +1175,81 @@ mod row_tests {
assert_eq!(ages, vec![30, 40, 50]);
}
+ #[tokio::test]
+ async fn
parquet_predicate_skips_row_groups_when_row_filtering_is_disabled() {
+ let fields = vec![field(0, "id", DataType::Int(IntType::new()))];
+ let schema = build_target_arrow_schema(&fields).unwrap();
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table_path = "memory:/parquet_pruning_only";
+ let bucket_path = format!("{table_path}/bucket-0");
+ let file_name = "part-0.parquet";
+ let file_path = format!("{bucket_path}/{file_name}");
+ let output = file_io.new_output(&file_path).unwrap();
+ let mut writer = create_format_writer(&output, schema.clone(), "zstd",
1, None, None, None)
+ .await
+ .unwrap();
+ writer
+ .write(
+ &RecordBatch::try_new(schema.clone(),
vec![Arc::new(Int32Array::from(vec![1, 2]))])
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ writer.flush().await.unwrap();
+ writer
+ .write(
+ &RecordBatch::try_new(schema,
vec![Arc::new(Int32Array::from(vec![100, 101]))])
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ let file_size = writer.close().await.unwrap().file_size as i64;
+
+ let schema_id = 1;
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(bucket_path)
+ .with_total_buckets(1)
+ .with_data_files(vec![data_file(file_name, file_size, 4,
schema_id)])
+ .build()
+ .unwrap();
+ let predicate = PredicateBuilder::new(&fields)
+ .greater_than("id", Datum::Int(100))
+ .unwrap();
+ let reader = DataFileReader::new(
+ file_io.clone(),
+ SchemaManager::new(file_io, table_path.to_string()),
+ schema_id,
+ fields.clone(),
+ fields,
+ vec![predicate],
+ )
+ .with_row_filter(false);
+ let batches = reader
+ .read(&[split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let ids = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .iter()
+ .copied()
+ })
+ .collect::<Vec<_>>();
+
+ assert_eq!(ids, vec![100, 101]);
+ }
+
/// Guard: projecting `_ROW_ID` together with a data predicate must fail
/// loudly rather than assign wrong row ids. `_ROW_ID` is materialized
/// positionally from post-filter batch row counts, so the readers'
residual
@@ -1253,7 +1340,7 @@ mod row_tests {
let read_type = vec![row_id];
// AlwaysTrue alone -> allowed.
assert!(
- DataFileReader::reject_row_id_with_predicate(&read_type,
&[Predicate::AlwaysTrue])
+ DataFileReader::reject_row_id_with_predicates(&read_type,
&[Predicate::AlwaysTrue])
.is_ok(),
"AlwaysTrue must not trip the _ROW_ID guard"
);
@@ -1262,7 +1349,7 @@ mod row_tests {
.greater_than("age", Datum::Int(1))
.unwrap();
assert!(
- DataFileReader::reject_row_id_with_predicate(&read_type,
&[filtering]).is_err(),
+ DataFileReader::reject_row_id_with_predicates(&read_type,
&[filtering]).is_err(),
"a row-filtering predicate must trip the _ROW_ID guard"
);
}
@@ -1336,6 +1423,23 @@ mod tests {
with_filter.has_row_filtering_predicate(),
"a real (non-AlwaysTrue) predicate is row-filtering"
);
+
+ let pruning = PredicateBuilder::new(&fields)
+ .equal("id", crate::spec::Datum::Int(10))
+ .unwrap();
+ let with_pruning = DataFileReader::new(
+ file_io,
+ schema_manager,
+ 1,
+ fields.clone(),
+ fields,
+ vec![pruning],
+ )
+ .with_row_filter(false);
+ assert!(
+ with_pruning.has_row_filtering_predicate(),
+ "a pruning predicate can skip physical rows"
+ );
}
struct MemOutputFile {
diff --git a/crates/paimon/src/table/format_table_read.rs
b/crates/paimon/src/table/format_table_read.rs
index 0d561ab1..0b5f72f2 100644
--- a/crates/paimon/src/table/format_table_read.rs
+++ b/crates/paimon/src/table/format_table_read.rs
@@ -38,6 +38,7 @@ pub(crate) struct FormatTableRead<'a> {
table: &'a Table,
read_type: Vec<DataField>,
data_predicates: Vec<Predicate>,
+ row_filter: bool,
limit: Option<usize>,
}
@@ -52,6 +53,7 @@ impl<'a> FormatTableRead<'a> {
table,
read_type,
data_predicates,
+ row_filter: true,
limit,
}
}
@@ -73,6 +75,11 @@ impl<'a> FormatTableRead<'a> {
self
}
+ pub(crate) fn with_row_filter(mut self, row_filter: bool) -> Self {
+ self.row_filter = row_filter;
+ self
+ }
+
pub(crate) fn to_arrow(
&self,
data_splits: &[DataSplit],
@@ -88,13 +95,13 @@ impl<'a> FormatTableRead<'a> {
let table_fields = self.table.schema().fields().to_vec();
let (data_table_fields, data_predicates) =
split_format_table_fields(&table_fields, &partition_keys,
&self.data_predicates);
-
let splits = data_splits.to_vec();
let file_io = self.table.file_io().clone();
let schema_manager = self.table.schema_manager().clone();
let schema_id = self.table.schema().id();
let mut remaining = self.limit;
let batch_size = Some(core_options.read_batch_size()?);
+ let row_filter = self.row_filter;
Ok(try_stream! {
for split in splits {
@@ -110,6 +117,7 @@ impl<'a> FormatTableRead<'a> {
data_read_type.clone(),
data_predicates.clone(),
)
+ .with_row_filter(row_filter)
.with_batch_size(batch_size)
.read(std::slice::from_ref(&split))?;
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index 4450b273..675f8ed4 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -55,6 +55,8 @@ pub(crate) struct KeyValueFileReader {
/// of a key survives); they are enforced by the post-merge residual
/// filter using the full `config.predicates` instead.
pushdown_predicates: Vec<Predicate>,
+ /// Whether predicates may remove rows before or after the key-value merge.
+ apply_row_filter: bool,
#[cfg(test)]
input_batch_sizes: Option<std::sync::Arc<std::sync::Mutex<Vec<usize>>>>,
}
@@ -118,11 +120,17 @@ impl KeyValueFileReader {
file_io,
config,
pushdown_predicates,
+ apply_row_filter: true,
#[cfg(test)]
input_batch_sizes: None,
}
}
+ pub(crate) fn with_row_filter(mut self, apply_row_filter: bool) -> Self {
+ self.apply_row_filter = apply_row_filter;
+ self
+ }
+
#[cfg(test)]
fn with_input_batch_sizes(
mut self,
@@ -235,6 +243,7 @@ impl KeyValueFileReader {
let residual_file_predicates =
(!self.config.predicates.is_empty()).then(||
crate::arrow::format::FilePredicates {
predicates: self.config.predicates.clone(),
+ apply_row_filter: self.apply_row_filter,
file_fields: self.config.table_fields.clone(),
});
let user_fields = crate::arrow::residual::widen_scan_fields(
@@ -325,6 +334,7 @@ impl KeyValueFileReader {
let table_options = self.config.table_options;
let pushdown_predicates = self.pushdown_predicates;
let residual_predicates = self.config.predicates;
+ let apply_row_filter = self.apply_row_filter;
let primary_keys = self.config.primary_keys;
let sequence_fields = self.config.sequence_fields;
let read_batch_size = self.config.read_batch_size;
@@ -368,6 +378,7 @@ impl KeyValueFileReader {
internal_read_type.clone(),
pushdown_predicates.clone(),
)
+ .with_row_filter(apply_row_filter)
.with_batch_size(Some(read_batch_size));
let stream = reader.read_single_file_stream(
@@ -421,15 +432,15 @@ impl KeyValueFileReader {
while let Some(batch) = merge_stream.next().await {
let batch = batch?;
- // Post-merge residual: enforce the FULL data predicate on
- // merged rows. PK conjuncts are also in this set (they
were
- // already pushed down pre-merge); re-evaluating them on
- // already-matching rows is a no-op and keeps one shared
- // evaluator instead of deriving a non-PK subset. Runs on
- // the merge-output batch (keys + values, including widened
- // predicate columns); the reorder below projects the
- // output back to read_type.
- let batch = if residual_predicates.is_empty() {
+ // When row filtering is enabled, the post-merge residual
+ // enforces the FULL data predicate on merged rows. PK
+ // conjuncts are also in this set (they were already pushed
+ // down pre-merge); re-evaluating them on already-matching
+ // rows is a no-op and keeps one shared evaluator instead
of
+ // deriving a non-PK subset. Runs on the merge-output batch
+ // (keys + values, including widened predicate columns);
the
+ // reorder below projects the output back to read_type.
+ let batch = if !apply_row_filter ||
residual_predicates.is_empty() {
batch
} else {
match crate::arrow::residual::evaluate_predicates_mask(
@@ -774,6 +785,42 @@ mod tests {
assert_eq!(int_column(&batches, "value"), vec![21]);
}
+ #[tokio::test]
+ async fn kv_read_row_filter_disabled_keeps_merged_rows() {
+ let file_io = test_file_io();
+ let table_path = "memory:/kv_pruning_only";
+ setup_dirs(&file_io, table_path).await;
+ let table = pk_table(&file_io, table_path, &[]);
+
+ write_commit(
+ &table,
+ &int_batch(vec![1, 2, 3], vec![Some(10), Some(20), Some(30)]),
+ )
+ .await;
+ write_commit(
+ &table,
+ &int_batch(vec![1, 2, 3], vec![Some(11), Some(21), Some(31)]),
+ )
+ .await;
+
+ let filter = PredicateBuilder::new(table.schema().fields())
+ .equal("value", Datum::Int(21))
+ .unwrap();
+ let mut builder = table.new_read_builder();
+ builder.with_filter(filter);
+ let plan = builder.new_scan().plan().await.unwrap();
+ let read = builder.new_read().unwrap().with_row_filter(false);
+ let batches = read
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(int_column(&batches, "id"), vec![1, 2, 3]);
+ assert_eq!(int_column(&batches, "value"), vec![11, 21, 31]);
+ }
+
/// Gap-A: the predicate column is NOT in the projection. The merge read
/// must widen internally, filter, then project back — output schema must
/// contain only the projected column.
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index d95c0104..5e4839a2 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -110,6 +110,22 @@ impl<'a> TableRead<'a> {
}
}
+ /// Configure whether data predicates may remove rows from emitted batches.
+ ///
+ /// When disabled, predicates remain available for conservative scan and
+ /// file-format pruning, and the caller must enforce them exactly. Merge
and
+ /// data-evolution readers do not push such predicates below
materialization.
+ pub fn with_row_filter(self, row_filter: bool) -> Self {
+ match self.0 {
+ TableReadKind::Paimon(read) => {
+ Self(TableReadKind::Paimon(read.with_row_filter(row_filter)))
+ }
+ TableReadKind::Format(read) => {
+ Self(TableReadKind::Format(read.with_row_filter(row_filter)))
+ }
+ }
+ }
+
/// Returns an [`ArrowRecordBatchStream`].
pub fn to_arrow(&self, data_splits: &[DataSplit]) ->
crate::Result<ArrowRecordBatchStream> {
match &self.0 {
@@ -158,6 +174,7 @@ struct PaimonTableRead<'a> {
table: &'a Table,
read_type: Vec<DataField>,
data_predicates: Vec<Predicate>,
+ row_filter: bool,
}
impl<'a> PaimonTableRead<'a> {
@@ -171,6 +188,7 @@ impl<'a> PaimonTableRead<'a> {
table,
read_type,
data_predicates,
+ row_filter: true,
}
}
@@ -189,9 +207,9 @@ impl<'a> PaimonTableRead<'a> {
self.table
}
- /// Set a filter predicate. Used conservatively for read-side pruning and
- /// enforced exactly by residual filtering on append, data-evolution, and
- /// primary-key merge read paths (see
+ /// Set a filter predicate. Used conservatively for read-side pruning and,
+ /// unless row filtering is disabled, enforced exactly by residual
filtering
+ /// on append, data-evolution, and primary-key merge read paths (see
/// [`ReadBuilder::with_filter`](crate::table::ReadBuilder::with_filter)
/// for per-format exceptions).
pub fn with_filter(mut self, filter: Predicate) -> Self {
@@ -204,6 +222,11 @@ impl<'a> PaimonTableRead<'a> {
self
}
+ fn with_row_filter(mut self, row_filter: bool) -> Self {
+ self.row_filter = row_filter;
+ self
+ }
+
/// Returns an [`ArrowRecordBatchStream`] for an incremental scan plan.
pub fn to_incremental_arrow(
&self,
@@ -286,6 +309,7 @@ impl<'a> PaimonTableRead<'a> {
read_type,
self.data_predicates.clone(),
)
+ .with_row_filter(self.row_filter)
.with_batch_size(Some(self.table.schema().core_options().read_batch_size()?));
let raw_stream = reader.read(&data_splits)?;
@@ -443,7 +467,8 @@ impl<'a> PaimonTableRead<'a> {
.collect(),
read_batch_size: core_options.read_batch_size()?,
},
- );
+ )
+ .with_row_filter(self.row_filter);
reader.read(splits)
}
@@ -466,6 +491,7 @@ impl<'a> PaimonTableRead<'a> {
core_options.blob_view_resolve_enabled(),
self.table.rest_env().cloned(),
)?
+ .with_row_filter(self.row_filter)
.with_batch_size(Some(core_options.read_batch_size()?));
reader.read(data_splits)
}
@@ -484,6 +510,7 @@ impl<'a> PaimonTableRead<'a> {
self.read_type().to_vec(),
self.data_predicates.clone(),
)
+ .with_row_filter(self.row_filter)
.with_batch_size(Some(self.table.schema().core_options().read_batch_size()?)))
}
}
@@ -641,6 +668,19 @@ mod tests {
assert!(!pk_split_needs_merge(&dv_compacted, true));
}
+ #[test]
+ fn test_row_filter_disabled_keeps_data_predicates() {
+ let table = query_auth_table();
+ let read = PaimonTableRead::new(
+ &table,
+ table.schema.fields().to_vec(),
+ vec![Predicate::AlwaysFalse],
+ )
+ .with_row_filter(false);
+
+ assert_eq!(read.data_predicates(), &[Predicate::AlwaysFalse]);
+ }
+
#[test]
fn test_rowkind_rejects_null_value_kind() {
let values = arrow_array::Int8Array::from(vec![Some(0), None]);
diff --git a/crates/paimon/src/table/vector_search_builder.rs
b/crates/paimon/src/table/vector_search_builder.rs
index 57cde2be..cbfdcfef 100644
--- a/crates/paimon/src/table/vector_search_builder.rs
+++ b/crates/paimon/src/table/vector_search_builder.rs
@@ -413,6 +413,7 @@ impl<'a> VectorSearchBuilder<'a> {
Some(filter) => {
let file_predicates = FilePredicates {
predicates: vec![filter.clone()],
+ apply_row_filter: true,
file_fields: self.table.schema().fields().to_vec(),
};
let residual_read_type = widen_scan_fields(&[],
Some(&file_predicates));
@@ -3407,6 +3408,7 @@ mod residual_positions_tests {
.unwrap();
FilePredicates {
predicates: vec![pred],
+ apply_row_filter: true,
file_fields: vec![id_field()],
}
}
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 01635ba6..0cd5e255 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1524,7 +1524,9 @@ SET 'paimon.scan.version' = '1';
RESET 'paimon.scan.version';
```
-Options prefixed with `paimon.` are handled by Paimon; all others are
delegated to DataFusion. Dynamic options are applied at table load time via
`table.copy_with_options()`.
+Quoted options prefixed with `paimon.` are handled as Paimon table options; all
+others are delegated to DataFusion. Dynamic table options are applied at table
+load time via `table.copy_with_options()`.
Example — enable BLOB descriptor mode:
