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 7cb6f8a0 fix(datafusion): keep format table partition columns out of
decoder filters (#811)
7cb6f8a0 is described below
commit 7cb6f8a00fdcf6fcf47edbf066f0c7e53de103c0
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Fri Sep 11 13:37:59 2026 +0800
fix(datafusion): keep format table partition columns out of decoder filters
(#811)
---
.../datafusion/src/physical_plan/scan.rs | 33 ++++-
.../tests/format_table_partition_filters.rs | 139 +++++++++++++++++++++
2 files changed, 171 insertions(+), 1 deletion(-)
diff --git a/crates/integrations/datafusion/src/physical_plan/scan.rs
b/crates/integrations/datafusion/src/physical_plan/scan.rs
index fe0a3859..d2fc56c3 100644
--- a/crates/integrations/datafusion/src/physical_plan/scan.rs
+++ b/crates/integrations/datafusion/src/physical_plan/scan.rs
@@ -50,7 +50,9 @@ use
datafusion::physical_plan::stream::RecordBatchStreamAdapter;
use datafusion::physical_plan::{DisplayAs, ExecutionPlan, Partitioning,
PlanProperties};
use futures::{FutureExt, StreamExt, TryStreamExt};
use paimon::arrow::ParquetReadBudget;
-use paimon::spec::{DataField, Datum, MergeEngine, Predicate, PredicateBuilder,
PredicateOperator};
+use paimon::spec::{
+ CoreOptions, DataField, Datum, MergeEngine, Predicate, PredicateBuilder,
PredicateOperator,
+};
use paimon::table::{ScanTrace, Table};
use paimon::DataSplit;
@@ -219,6 +221,31 @@ struct RuntimeDecoderFilterPlan {
datafusion_filters: Vec<Arc<dyn PhysicalExpr>>,
}
+/// Whether a conjunct reads a partition column whose values the data files do
not hold.
+///
+/// A format table keeps partition values in its directory names only, so a
decoder filter on such
+/// a column would read it as missing and drop every row. Those conjuncts are
left to the runtime
+/// filters, which run on batches with the partition columns filled in.
+fn reads_partition_column_absent_from_files(
+ conjunct: &Arc<dyn PhysicalExpr>,
+ table: &Table,
+ case_sensitive: bool,
+) -> bool {
+ if !CoreOptions::new(table.schema().options()).is_format_table() {
+ return false;
+ }
+ let partition_keys = table.schema().partition_keys();
+ collect_columns(conjunct).iter().any(|column| {
+ partition_keys.iter().any(|key| {
+ if case_sensitive {
+ key == column.name()
+ } else {
+ key.eq_ignore_ascii_case(column.name())
+ }
+ })
+ })
+}
+
fn partition_runtime_decoder_filters(
decoder_filters: &[Arc<dyn PhysicalExpr>],
fields: &[DataField],
@@ -1040,6 +1067,10 @@ impl ExecutionPlan for PaimonTableScan {
conjunct,
self.table.schema().fields(),
self.case_sensitive,
+ ) && !reads_partition_column_absent_from_files(
+ conjunct,
+ &self.table,
+ self.case_sensitive,
)
})
.cloned(),
diff --git
a/crates/integrations/datafusion/tests/format_table_partition_filters.rs
b/crates/integrations/datafusion/tests/format_table_partition_filters.rs
new file mode 100644
index 00000000..fc302776
--- /dev/null
+++ b/crates/integrations/datafusion/tests/format_table_partition_filters.rs
@@ -0,0 +1,139 @@
+// 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.
+
+//! Filters on the partition columns of a `type=format-table` table.
+//!
+//! A format table keeps partition values in its directory names only; the data
+//! files do not hold those columns. A filter the scan cannot turn into a
+//! partition predicate still has to see the value from the directory name.
+
+use std::path::Path;
+use std::sync::Arc;
+
+use arrow_array::{Int64Array, RecordBatch};
+use arrow_schema::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
+use paimon::catalog::{Catalog, FileSystemCatalog, Identifier};
+use paimon::spec::{BigIntType, BooleanType, DataType, Schema, VarCharType};
+use paimon::{CatalogOptions, Options};
+use paimon_datafusion::SQLContext;
+use parquet::arrow::ArrowWriter;
+use tempfile::TempDir;
+
+const DATABASE: &str = "test_db";
+const TABLE: &str = "events";
+
+/// A format table partitioned by `(dt, active)` with one data file per
partition.
+///
+/// The warehouse is given as `file:/...`, the form in which listed file paths
come back,
+/// so that a scan listing from the table root can relate each file to its
partition
+/// directory.
+async fn setup_table() -> (TempDir, SQLContext) {
+ let tmp = TempDir::new().expect("Failed to create temp dir");
+ let mut options = Options::new();
+ options.set(
+ CatalogOptions::WAREHOUSE,
+ format!("file:{}", tmp.path().display()),
+ );
+ let catalog = Arc::new(FileSystemCatalog::new(options).expect("Failed to
create catalog"));
+ catalog
+ .create_database(DATABASE, false, Default::default())
+ .await
+ .expect("CREATE DATABASE failed");
+ let schema = Schema::builder()
+ .column("dt", DataType::VarChar(VarCharType::new(32).unwrap()))
+ .column("active", DataType::Boolean(BooleanType::new()))
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .partition_keys(vec!["dt".to_string(), "active".to_string()])
+ .option("type", "format-table")
+ .option("file.format", "parquet")
+ .build()
+ .unwrap();
+ catalog
+ .create_table(&Identifier::new(DATABASE, TABLE), schema, false)
+ .await
+ .expect("CREATE TABLE failed");
+ let table_dir = tmp.path().join(format!("{DATABASE}.db")).join(TABLE);
+ for (dt, active, id) in [("a", true, 1), ("b", false, 2)] {
+ write_ids(&table_dir.join(format!("dt={dt}/active={active}")), &[id]);
+ }
+ let mut context = SQLContext::new();
+ context.register_catalog("paimon", catalog).await.unwrap();
+ (tmp, context)
+}
+
+fn write_ids(directory: &Path, ids: &[i64]) {
+ std::fs::create_dir_all(directory).unwrap();
+ let schema = Arc::new(ArrowSchema::new(vec![Field::new(
+ "id",
+ ArrowDataType::Int64,
+ true,
+ )]));
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int64Array::from(ids.to_vec()))],
+ )
+ .unwrap();
+ let file =
std::fs::File::create(directory.join("part-0.parquet")).unwrap();
+ let mut writer = ArrowWriter::try_new(file, schema, None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+}
+
+async fn ids(context: &SQLContext, sql: &str) -> Vec<i64> {
+ let mut ids = Vec::new();
+ for batch in context.sql(sql).await.unwrap().collect().await.unwrap() {
+ let values = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ ids.extend(values.iter().flatten());
+ }
+ ids.sort_unstable();
+ ids
+}
+
+/// A filter on a partition column that is not a partition predicate, such as a
+/// bare boolean column (DataFusion simplifies `active = true` to `active`) or
a
+/// function over the column, must be evaluated with the value from the
+/// directory name rather than dropping every row as a missing column.
+#[tokio::test]
+async fn test_filter_on_a_partition_column_reads_the_directory_value() {
+ let (_tmp, context) = setup_table().await;
+
+ let mut mismatches = Vec::new();
+ for (predicate, expected) in [
+ ("dt IN ('a', 'b')", vec![1, 2]),
+ ("active", vec![1]),
+ ("active = true", vec![1]),
+ ("NOT active", vec![2]),
+ ("upper(dt) = 'B'", vec![2]),
+ ("concat(dt, '-') = 'a-'", vec![1]),
+ ] {
+ let actual = ids(
+ &context,
+ &format!("SELECT id FROM paimon.{DATABASE}.{TABLE} WHERE
{predicate}"),
+ )
+ .await;
+ if actual != expected {
+ mismatches.push(format!(
+ "{predicate}: expected {expected:?}, got {actual:?}"
+ ));
+ }
+ }
+ assert!(mismatches.is_empty(), "{mismatches:#?}");
+}
