This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6067-5d59317630e19d22cc9be5dbdae59ae1b87b4f41 in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 29adcffcf32d7bd135c09187d2569891e9dbc25b Author: Ping Zhang <[email protected]> AuthorDate: Thu Sep 24 02:31:26 2026 +0000 fix: preserve Parquet conversion errors during join filtering (#6067) * fix: preserve Parquet conversion errors during join filtering * fix: preserve column names in runtime filter checks * fix: require resolved columns for Parquet runtime filtering --- docs/source/user-guide/latest/tuning.md | 8 + .../operators/dynamic_filter/join/tests.rs | 3 + .../dynamic_filter/join/tests/schema_errors.rs | 234 +++++++++++++++++++++ .../join/tests/schema_errors/partition_columns.rs | 177 ++++++++++++++++ .../dynamic_filter/join/tests/timestamp_errors.rs | 139 ++++++++++++ .../operators/dynamic_filter/parquet_reader.rs | 76 ++++++- .../parquet_reader/schema_adapter.rs | 109 ++++++++++ .../parquet_reader/schema_adapter/tests.rs | 132 ++++++++++++ .../schema_adapter/tests/resolution.rs | 133 ++++++++++++ .../org/apache/comet/exec/CometJoinSuite.scala | 58 +++++ 10 files changed, 1066 insertions(+), 3 deletions(-) diff --git a/docs/source/user-guide/latest/tuning.md b/docs/source/user-guide/latest/tuning.md index 368f46ece6..5855d6cdc0 100644 --- a/docs/source/user-guide/latest/tuning.md +++ b/docs/source/user-guide/latest/tuning.md @@ -381,6 +381,14 @@ projects the file schema. The original null checks and residual runtime filter r The original join still verifies matches, including any hash collisions admitted by the filter. Standalone projections, other filter expressions, and limits prevent reader attachment. +To preserve schema-conversion and timestamp-overflow errors, runtime reader pruning is disabled for +each file whose projected or statically filtered columns require schema adaptations beyond direct +column mappings or literal values. This conservative check also disables reader pruning for allowed +`INT32` to `BIGINT` promotion and for projecting a subset of a struct's fields, even when those +adaptations cannot fail. Nested column pruning still reads only the requested struct fields. Scans +with supplied file statistics also skip reader attachment. These cases still use runtime filtering +on decoded batches. + Filters stay within the task's native plan and do not propagate across Spark exchanges or JVM/Arrow boundaries. A shuffled hash join can still filter probe batches after shuffle, but it cannot send its filter back to an earlier scan stage. Compare the [runtime-filter and scan metrics](metrics.md#hash-joins) diff --git a/native/core/src/execution/operators/dynamic_filter/join/tests.rs b/native/core/src/execution/operators/dynamic_filter/join/tests.rs index aa64656f73..5e2a4f3923 100644 --- a/native/core/src/execution/operators/dynamic_filter/join/tests.rs +++ b/native/core/src/execution/operators/dynamic_filter/join/tests.rs @@ -15,6 +15,9 @@ // specific language governing permissions and limitations // under the License. +mod schema_errors; +mod timestamp_errors; + use super::*; use std::fmt::Display; use std::hash::{Hash, Hasher}; diff --git a/native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors.rs b/native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors.rs new file mode 100644 index 0000000000..8fa236f23d --- /dev/null +++ b/native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors.rs @@ -0,0 +1,234 @@ +// 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 super::*; + +mod partition_columns; + +fn write_file(payload_type: &DataType, keys: std::ops::Range<i32>) -> tempfile::NamedTempFile { + let schema = Arc::new(Schema::new(vec![ + Field::new("key", DataType::Int32, false), + Field::new("payload", payload_type.clone(), false), + ])); + let values = Int32Array::from_iter_values(keys); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(values.clone()), + cast(&values, payload_type).unwrap(), + ], + ) + .unwrap(); + let file = tempfile::NamedTempFile::new().unwrap(); + let mut writer = ArrowWriter::try_new( + file.reopen().unwrap(), + schema, + Some( + WriterProperties::builder() + .set_dictionary_enabled(false) + .build(), + ), + ) + .unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + file +} + +fn scan( + files: &[&tempfile::NamedTempFile], + project_payload: bool, + filters: Option<Vec<Arc<dyn PhysicalExpr>>>, + allow_type_promotion: bool, + session: &Arc<SessionContext>, +) -> Arc<DataSourceExec> { + let data_schema = Arc::new(Schema::new(vec![ + Field::new("key", DataType::Int32, false), + Field::new("payload", DataType::Int64, false), + ])); + let projection = if project_payload { vec![0, 1] } else { vec![0] }; + let required_schema = Arc::new(data_schema.project(&projection).unwrap()); + init_datasource_exec( + required_schema, + Some(data_schema), + None, + ObjectStoreUrl::local_filesystem(), + ObjectStoreBackend::Local, + vec![files + .iter() + .map(|file| { + PartitionedFile::from_path(file.path().to_string_lossy().into_owned()).unwrap() + }) + .collect()], + Some(projection), + filters, + None, + "UTC", + true, + false, + allow_type_promotion, + false, + session, + false, + false, + false, + ) + .unwrap() +} + +fn filtered_join( + scan: Arc<DataSourceExec>, + enabled: bool, + session: &Arc<SessionContext>, +) -> Arc<dyn ExecutionPlan> { + let build = memory_exec(vec![RecordBatch::try_new( + Arc::new(Schema::new(vec![Field::new("key", DataType::Int32, false)])), + vec![Arc::new(Int32Array::from(vec![0]))], + ) + .unwrap()]); + let join = single_key_join_plans(build, scan, PartitionMode::Partitioned); + PhysicalPlanner::apply_join_dynamic_filter( + Arc::new(join), + enabled, + session.copied_config().options(), + ) + .unwrap() +} + +/// A nonmatching file still has to raise its projected payload conversion error. +/// Spark 3 rejects INT32 -> BIGINT unless type promotion is explicitly allowed. +#[tokio::test] +async fn join_reader_filter_preserves_schema_conversion_error() { + let compatible = write_file(&DataType::Int64, 0..4); + let incompatible = write_file(&DataType::Int32, 100..104); + for row_filter in [false, true] { + for enabled in [false, true] { + let mut config = SessionConfig::new() + .with_target_partitions(1) + .with_parquet_page_index_pruning(false); + config.options_mut().execution.parquet.pushdown_filters = row_filter; + let session = Arc::new(SessionContext::new_with_config(config)); + let scan = scan(&[&compatible, &incompatible], true, None, false, &session); + let plan = filtered_join(scan, enabled, &session); + let result = collect(plan, session.task_ctx()).await; + let error = result.expect_err(&format!( + "projected INT32 -> BIGINT must fail: enabled={enabled}, row_filter={row_filter}" + )); + assert!( + error + .to_string() + .contains("Parquet column cannot be converted"), + "enabled={enabled}, row_filter={row_filter}: {error}" + ); + } + } +} + +/// Supplied file statistics allow pruning before the file's schema adapter runs. +#[tokio::test] +async fn join_reader_filter_preserves_schema_error_with_file_statistics() { + use datafusion::common::stats::Precision; + use datafusion::common::{ScalarValue, Statistics}; + + let compatible = write_file(&DataType::Int64, 0..4); + let incompatible = write_file(&DataType::Int32, 100..104); + for enabled in [false, true] { + let mut config = SessionConfig::new() + .with_target_partitions(1) + .with_parquet_page_index_pruning(false); + config.options_mut().execution.parquet.pushdown_filters = false; + let session = Arc::new(SessionContext::new_with_config(config)); + let scan = scan(&[&compatible, &incompatible], true, None, false, &session); + let (config, _) = scan.downcast_to_file_source::<ParquetSource>().unwrap(); + let mut config = config.clone(); + config.file_groups = vec![config.file_groups[0] + .files() + .iter() + .cloned() + .zip([0, 100]) + .map(|(file, first_key)| { + let mut statistics = Statistics::new_unknown(&scan.schema()); + statistics.num_rows = Precision::Exact(4); + statistics.column_statistics[0].min_value = + Precision::Exact(ScalarValue::Int32(Some(first_key))); + statistics.column_statistics[0].max_value = + Precision::Exact(ScalarValue::Int32(Some(first_key + 3))); + statistics.column_statistics[0].null_count = Precision::Exact(0); + file.with_statistics(Arc::new(statistics)) + }) + .collect::<Vec<_>>() + .into()]; + let scan = Arc::new(scan.as_ref().clone().with_data_source(Arc::new(config))); + let plan = filtered_join(scan, enabled, &session); + let error = collect(plan, session.task_ctx()).await.expect_err(&format!( + "file statistics must preserve the schema error: enabled={enabled}" + )); + assert!( + error + .to_string() + .contains("Parquet column cannot be converted"), + "enabled={enabled}: {error}" + ); + } +} + +/// Valid conversions and unread mismatched columns must remain readable. Spark's +/// static predicates and empty files can also legitimately avoid conversion errors. +#[tokio::test] +async fn join_reader_schema_guard_preserves_readable_cases() { + let compatible = write_file(&DataType::Int64, 0..4); + for scenario in ["empty", "static", "unprojected", "allowed"] { + let incompatible = write_file( + &DataType::Int32, + if scenario == "empty" { + 100..100 + } else { + 100..104 + }, + ); + for enabled in [false, true] { + let mut config = SessionConfig::new() + .with_target_partitions(1) + .with_parquet_page_index_pruning(false); + config.options_mut().execution.parquet.pushdown_filters = false; + let session = Arc::new(SessionContext::new_with_config(config)); + let filters = (scenario == "static").then(|| { + vec![Arc::new(BinaryExpr::new( + Arc::new(Column::new("key", 0)), + Operator::Lt, + lit(100_i32), + )) as Arc<dyn PhysicalExpr>] + }); + let scan = scan( + &[&compatible, &incompatible], + scenario != "unprojected", + filters, + scenario == "allowed", + &session, + ); + let plan = filtered_join(scan, enabled, &session); + let output = collect(plan, session.task_ctx()) + .await + .unwrap_or_else(|error| panic!("scenario={scenario}, enabled={enabled}: {error}")); + assert_eq!( + row_count(&output), + 1, + "scenario={scenario}, enabled={enabled}" + ); + } + } +} diff --git a/native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors/partition_columns.rs b/native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors/partition_columns.rs new file mode 100644 index 0000000000..3a4d55ae06 --- /dev/null +++ b/native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors/partition_columns.rs @@ -0,0 +1,177 @@ +// 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 super::*; +use datafusion::common::ScalarValue; +use datafusion::physical_expr::utils::collect_columns; + +fn partitioned_session() -> Arc<SessionContext> { + let mut config = SessionConfig::new() + .with_target_partitions(1) + .with_parquet_page_index_pruning(false); + // Isolate row-group pruning from the residual batch filter. + config.options_mut().execution.parquet.pushdown_filters = false; + Arc::new(SessionContext::new_with_config(config)) +} + +fn partitioned_scan( + files: &[(&tempfile::NamedTempFile, i32)], + projection: Vec<usize>, + filters: Option<Vec<Arc<dyn PhysicalExpr>>>, + session: &Arc<SessionContext>, +) -> Arc<DataSourceExec> { + let file_schema = Arc::new(Schema::new(vec![ + Field::new("key", DataType::Int32, false), + Field::new("payload", DataType::Int64, false), + ])); + let partition_field = Field::new("part", DataType::Int32, false); + let table_schema = Schema::new(vec![ + file_schema.field(0).clone(), + file_schema.field(1).clone(), + partition_field.clone(), + ]); + init_datasource_exec( + Arc::new(table_schema.project(&projection).unwrap()), + Some(file_schema), + Some(Arc::new(Schema::new(vec![partition_field]))), + ObjectStoreUrl::local_filesystem(), + ObjectStoreBackend::Local, + vec![files + .iter() + .map(|(file, partition)| { + let mut file = + PartitionedFile::from_path(file.path().to_string_lossy().into_owned()).unwrap(); + file.partition_values = vec![ScalarValue::Int32(Some(*partition))]; + file + }) + .collect()], + Some(projection), + filters, + None, + "UTC", + true, + false, + false, + false, + session, + false, + false, + false, + ) + .unwrap() +} + +#[tokio::test] +async fn projected_partition_preserves_payload_conversion_error() { + let compatible = write_file(&DataType::Int64, 0..4); + let incompatible = write_file(&DataType::Int32, 100..104); + for enabled in [false, true] { + let session = partitioned_session(); + let scan = partitioned_scan( + &[(&compatible, 7), (&incompatible, 8)], + vec![0, 2, 1], + None, + &session, + ); + let result = collect(filtered_join(scan, enabled, &session), session.task_ctx()).await; + let error = result.expect_err("the nonmatching partition still has an invalid payload"); + assert!( + error + .to_string() + .contains("Parquet column cannot be converted"), + "enabled={enabled}: {error}" + ); + } +} + +#[tokio::test] +async fn projected_partition_keeps_runtime_reader_pruning() { + let matching = write_file(&DataType::Int64, 0..4); + let nonmatching = write_file(&DataType::Int64, 100..104); + let mut outputs = Vec::new(); + for enabled in [false, true] { + let session = partitioned_session(); + let scan = partitioned_scan( + &[(&matching, 7), (&nonmatching, 8)], + vec![0, 2, 1], + None, + &session, + ); + let plan = filtered_join(Arc::clone(&scan), enabled, &session); + let output = collect(Arc::clone(&plan), session.task_ctx()) + .await + .unwrap(); + assert_eq!(row_count(&output), 1, "enabled={enabled}"); + let batch = output.iter().find(|batch| batch.num_rows() > 0).unwrap(); + assert_eq!(batch.schema().field(2).name(), "part"); + assert_eq!( + ScalarValue::try_from_array(batch.column(2), 0).unwrap(), + ScalarValue::Int32(Some(7)) + ); + assert_eq!( + pruning_metric(&scan, "row_groups_pruned_statistics"), + usize::from(enabled), + "partition literals must not disable safe runtime reader pruning" + ); + if enabled { + assert_eq!(metric(&plan, "dynamic_filter_join_filters_attached"), 1); + } + outputs.push(batches_to_sort_string(&output)); + } + assert_eq!(outputs[0], outputs[1]); +} + +#[tokio::test] +async fn unprojected_partition_predicate_keeps_runtime_reader_pruning() { + let matching = write_file(&DataType::Int64, 0..4); + let nonmatching = write_file(&DataType::Int64, 100..104); + let mut outputs = Vec::new(); + let mut pruned_groups = Vec::new(); + for enabled in [false, true] { + let session = partitioned_session(); + let scan = partitioned_scan( + &[(&matching, 7), (&nonmatching, 7), (&matching, 8)], + vec![0, 1], + Some(vec![Arc::new(BinaryExpr::new( + Arc::new(Column::new("part", 2)), + Operator::Eq, + lit(7_i32), + ))]), + &session, + ); + assert!(scan.schema().index_of("part").is_err()); + let (_, source) = scan.downcast_to_file_source::<ParquetSource>().unwrap(); + assert!(collect_columns(&source.filter().unwrap()).contains(&Column::new("part", 2))); + let plan = filtered_join(Arc::clone(&scan), enabled, &session); + let output = collect(Arc::clone(&plan), session.task_ctx()) + .await + .unwrap(); + // The static predicate excludes partition 8 even though its key matches. + assert_eq!(row_count(&output), 1, "enabled={enabled}"); + if enabled { + assert_eq!(metric(&plan, "dynamic_filter_join_filters_attached"), 1); + } + pruned_groups.push(pruning_metric(&scan, "row_groups_pruned_statistics")); + outputs.push(batches_to_sort_string(&output)); + } + assert_eq!(outputs[0], outputs[1]); + assert_eq!( + pruned_groups[1], + pruned_groups[0] + 1, + "runtime pruning must still exclude the nonmatching row group" + ); +} diff --git a/native/core/src/execution/operators/dynamic_filter/join/tests/timestamp_errors.rs b/native/core/src/execution/operators/dynamic_filter/join/tests/timestamp_errors.rs new file mode 100644 index 0000000000..2da9af30cb --- /dev/null +++ b/native/core/src/execution/operators/dynamic_filter/join/tests/timestamp_errors.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. + +use super::*; +use arrow::array::{StructArray, TimestampMicrosecondArray, TimestampMillisecondArray}; + +fn timestamp_payload(values: ArrayRef, nested: bool) -> ArrayRef { + if nested { + Arc::new(StructArray::new( + vec![Field::new("ts", values.data_type().clone(), false)].into(), + vec![values], + None, + )) + } else { + values + } +} + +async fn assert_timestamp_overflow_preserved(nested: bool) { + let mut config = SessionConfig::new() + .with_target_partitions(1) + .with_parquet_page_index_pruning(false); + // Isolate runtime row-group pruning from page and row filtering. + config.options_mut().execution.parquet.pushdown_filters = false; + let session = Arc::new(SessionContext::new_with_config(config)); + let payload = timestamp_payload( + Arc::new(TimestampMillisecondArray::from_iter_values((0..200).map( + |key| { + if key < 100 { + 0 + } else { + i64::MAX / 1_000 + 1 + } + }, + ))), + nested, + ); + let key_field = Field::new("key", DataType::Int32, false); + let physical_schema = Arc::new(Schema::new(vec![ + key_field.clone(), + Field::new("payload", payload.data_type().clone(), false), + ])); + let batch = RecordBatch::try_new( + Arc::clone(&physical_schema), + vec![Arc::new(Int32Array::from_iter_values(0..200)), payload], + ) + .unwrap(); + let file = tempfile::NamedTempFile::new().unwrap(); + let properties = WriterProperties::builder() + .set_max_row_group_row_count(Some(100)) + .set_statistics_enabled(EnabledStatistics::Chunk) + .set_dictionary_enabled(false) + .build(); + let mut writer = + ArrowWriter::try_new(file.reopen().unwrap(), physical_schema, Some(properties)).unwrap(); + writer.write(&batch).unwrap(); + let metadata = writer.close().unwrap(); + assert_eq!(metadata.num_row_groups(), 2); + + // The build matches only the valid first row group. Spark's logical schema + // requests microseconds for the physical millisecond timestamp payload. + let logical_payload = timestamp_payload( + Arc::new(TimestampMicrosecondArray::from(Vec::<i64>::new())), + nested, + ); + let logical_schema = Arc::new(Schema::new(vec![ + key_field, + Field::new("payload", logical_payload.data_type().clone(), false), + ])); + for enabled in [false, true] { + // The two plans read the same file with fresh scan execution state. + let scan = init_datasource_exec( + Arc::clone(&logical_schema), + Some(Arc::clone(&logical_schema)), + None, + ObjectStoreUrl::local_filesystem(), + ObjectStoreBackend::Local, + vec![vec![PartitionedFile::from_path( + file.path().to_string_lossy().into_owned(), + ) + .unwrap()]], + Some(vec![0, 1]), + None, + None, + "UTC", + true, + false, + false, + false, + &session, + false, + false, + false, + ) + .unwrap(); + let join = single_key_join_plans( + input(vec![Some(0)], &DataType::Int32, 0), + scan, + PartitionMode::Partitioned, + ); + let plan = PhysicalPlanner::apply_join_dynamic_filter( + Arc::new(join), + enabled, + session.copied_config().options(), + ) + .unwrap(); + let error = collect(plan, session.task_ctx()) + .await + .expect_err("join runtime filtering must preserve the projected timestamp error"); + assert!( + error.to_string().to_lowercase().contains("overflow"), + "enabled={enabled}, nested={nested}: {error}" + ); + } +} + +#[tokio::test] +async fn reader_filter_preserves_timestamp_overflow() { + assert_timestamp_overflow_preserved(false).await; +} + +#[tokio::test] +async fn reader_filter_preserves_nested_timestamp_overflow() { + assert_timestamp_overflow_preserved(true).await; +} diff --git a/native/core/src/execution/operators/dynamic_filter/parquet_reader.rs b/native/core/src/execution/operators/dynamic_filter/parquet_reader.rs index be3eb1c40e..e97c8a0e65 100644 --- a/native/core/src/execution/operators/dynamic_filter/parquet_reader.rs +++ b/native/core/src/execution/operators/dynamic_filter/parquet_reader.rs @@ -21,17 +21,22 @@ use std::sync::Arc; use datafusion::common::config::ConfigOptions; use datafusion::common::Result; -use datafusion::datasource::physical_plan::ParquetSource; +use datafusion::datasource::physical_plan::{FileSource, ParquetSource}; use datafusion::datasource::source::DataSourceExec; use datafusion::logical_expr::Operator; use datafusion::physical_expr::expressions::{ BinaryExpr, Column, DynamicFilterPhysicalExpr, IsNotNullExpr, }; +use datafusion::physical_expr::utils::collect_columns; use datafusion::physical_expr::PhysicalExpr; use datafusion::physical_plan::ExecutionPlan; use super::super::CometFilterExec; +mod schema_adapter; + +use schema_adapter::RuntimeFilterSchemaAdapterFactory; + /// Recognize only direct-column null checks joined by AND, without evaluating /// or changing the predicate. Every accepted leaf is deterministic, infallible, /// and only discards rows, so reader pruning cannot suppress expression errors @@ -97,10 +102,64 @@ pub(super) fn try_attach_parquet_reader_filter( ); return Ok(None); }; - if scan.downcast_to_file_source::<ParquetSource>().is_none() { + let Some((file_config, source)) = scan.downcast_to_file_source::<ParquetSource>() else { log::debug!("Join dynamic filter reader pushdown skipped: probe is not Parquet"); return Ok(None); + }; + // File statistics can discard a file before its schema adapter is created. + // Retain the decoded-batch filter when per-file checks cannot guard pruning. + if file_config + .file_groups + .iter() + .flat_map(|group| group.files()) + .any(|file| file.statistics.is_some()) + { + log::debug!("Join dynamic filter reader pushdown skipped: supplied file statistics"); + return Ok(None); + } + let Some(adapter_factory) = &file_config.expr_adapter_factory else { + log::debug!("Join dynamic filter reader pushdown skipped: missing schema adapter factory"); + return Ok(None); + }; + let table_schema = source.table_schema(); + let file_column_count = table_schema.file_schema().fields().len(); + let projection = source + .projection() + .map(|projection| projection.column_indices()) + .unwrap_or_else(|| (0..file_column_count).collect()); + let mut read_columns = Vec::new(); + for index in projection { + let Some(field) = table_schema.table_schema().fields().get(index) else { + log::debug!( + "Join dynamic filter reader pushdown skipped: projected column index {index} is outside the table schema" + ); + return Ok(None); + }; + // Partition columns become literals before the per-file adapter is created. + if index < file_column_count { + read_columns.push(Column::new(field.name(), index)); + } + } + if let Some(filter) = source.filter() { + // Adapters resolve predicate columns by name and can repair stale indices. + // Keep that identity, including when excluding partition-column literals. + read_columns.extend(collect_columns(&filter).into_iter().filter(|column| { + !table_schema + .table_partition_cols() + .iter() + .any(|field| field.name() == column.name()) + })); } + read_columns.sort_unstable_by(|a, b| { + a.index() + .cmp(&b.index()) + .then_with(|| a.name().cmp(b.name())) + }); + read_columns.dedup(); + let adapter_factory = Arc::new(RuntimeFilterSchemaAdapterFactory::new( + Arc::clone(adapter_factory), + read_columns, + )); let predicate: Arc<dyn PhysicalExpr> = predicate; let propagation = match scan @@ -119,7 +178,18 @@ pub(super) fn try_attach_parquet_reader_filter( log::debug!("Join dynamic filter reader pushdown skipped: Parquet declined the predicate"); return Ok(None); }; - Ok(Some(Arc::new(scan.clone().with_data_source(data_source)))) + let filtered = scan.clone().with_data_source(data_source); + let Some((file_config, _)) = filtered.downcast_to_file_source::<ParquetSource>() else { + log::debug!( + "Join dynamic filter reader pushdown skipped: pushed-down scan is no longer Parquet" + ); + return Ok(None); + }; + let mut file_config = file_config.clone(); + file_config.expr_adapter_factory = Some(adapter_factory); + Ok(Some(Arc::new( + filtered.with_data_source(Arc::new(file_config)), + ))) } #[cfg(test)] diff --git a/native/core/src/execution/operators/dynamic_filter/parquet_reader/schema_adapter.rs b/native/core/src/execution/operators/dynamic_filter/parquet_reader/schema_adapter.rs new file mode 100644 index 0000000000..69163c0d0e --- /dev/null +++ b/native/core/src/execution/operators/dynamic_filter/parquet_reader/schema_adapter.rs @@ -0,0 +1,109 @@ +// 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. + +//! Preserve per-file conversion errors when runtime reader filters are attached. + +use std::sync::Arc; + +use arrow::datatypes::SchemaRef; +use datafusion::common::tree_node::{Transformed, TransformedResult, TreeNode}; +use datafusion::common::Result; +use datafusion::physical_expr::expressions::{lit, Column, DynamicFilterPhysicalExpr, Literal}; +use datafusion::physical_expr::PhysicalExpr; +use datafusion::physical_expr_adapter::{PhysicalExprAdapter, PhysicalExprAdapterFactory}; + +/// Wrap one execution's scan adapter factory, checking conversions separately for each file. +#[derive(Debug)] +pub(super) struct RuntimeFilterSchemaAdapterFactory { + inner: Arc<dyn PhysicalExprAdapterFactory>, + read_columns: Vec<Column>, +} + +impl RuntimeFilterSchemaAdapterFactory { + pub(super) fn new( + inner: Arc<dyn PhysicalExprAdapterFactory>, + read_columns: Vec<Column>, + ) -> Self { + Self { + inner, + read_columns, + } + } +} + +impl PhysicalExprAdapterFactory for RuntimeFilterSchemaAdapterFactory { + fn create( + &self, + logical_schema: SchemaRef, + physical_schema: SchemaRef, + ) -> Result<Arc<dyn PhysicalExprAdapter>> { + let inner = self + .inner + .create(logical_schema, Arc::clone(&physical_schema))?; + // DataFusion adapts predicates before projections and row-group pruning. + // Direct remapping and missing/default literals are safe. Other adaptations + // can reject nonempty batches or overflow, so let normal decoding run first. + // Spark's adapter can leave unresolved columns unchanged. Require rewritten + // column names to resolve in the original physical schema; case and field-ID + // remapping restore those names before returning the expression. + // Safe-adaptation and matching-schema optimizations are tracked in + // https://github.com/apache/datafusion-comet/issues/6123. + let allow_runtime_filter = + self.read_columns + .iter() + .all(|column| match inner.rewrite(Arc::new(column.clone())) { + Ok(expr) if expr.is::<Literal>() => true, + Ok(expr) => expr + .downcast_ref::<Column>() + .is_some_and(|column| physical_schema.index_of(column.name()).is_ok()), + Err(_) => false, + }); + Ok(Arc::new(RuntimeFilterSchemaAdapter { + inner, + allow_runtime_filter, + })) + } +} + +#[derive(Debug)] +struct RuntimeFilterSchemaAdapter { + inner: Arc<dyn PhysicalExprAdapter>, + allow_runtime_filter: bool, +} + +impl PhysicalExprAdapter for RuntimeFilterSchemaAdapter { + fn rewrite(&self, expr: Arc<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>> { + let expr = if self.allow_runtime_filter { + expr + } else { + expr.transform_down(|expr| { + if expr.is::<DynamicFilterPhysicalExpr>() { + Ok(Transformed::yes(lit(true))) + } else { + Ok(Transformed::no(expr)) + } + }) + .data()? + }; + // Keep static predicates and normal error timing: empty or statically + // excluded files must not acquire eager conversion failures. + self.inner.rewrite(expr) + } +} + +#[cfg(test)] +mod tests; diff --git a/native/core/src/execution/operators/dynamic_filter/parquet_reader/schema_adapter/tests.rs b/native/core/src/execution/operators/dynamic_filter/parquet_reader/schema_adapter/tests.rs new file mode 100644 index 0000000000..85b5bbe588 --- /dev/null +++ b/native/core/src/execution/operators/dynamic_filter/parquet_reader/schema_adapter/tests.rs @@ -0,0 +1,132 @@ +// 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. + +mod resolution; + +use super::*; +use arrow::array::{BooleanArray, Int32Array, RecordBatch}; +use arrow::datatypes::{DataType, Field, Schema}; +use datafusion::logical_expr::Operator; +use datafusion::physical_expr::expressions::BinaryExpr; +use datafusion::physical_expr_adapter::DefaultPhysicalExprAdapterFactory; + +fn schemas_and_batch() -> (SchemaRef, RecordBatch) { + let logical_schema = Arc::new(Schema::new(vec![ + Field::new("key", DataType::Int32, false), + Field::new("payload", DataType::Int64, false), + ])); + let physical_schema = Arc::new(Schema::new(vec![ + Field::new("key", DataType::Int32, false), + Field::new("payload", DataType::Int32, false), + ])); + let batch = RecordBatch::try_new( + physical_schema, + vec![ + Arc::new(Int32Array::from(vec![1, 2])), + Arc::new(Int32Array::from(vec![10, 20])), + ], + ) + .unwrap(); + (logical_schema, batch) +} + +#[test] +fn conversion_guard_resolves_static_columns_by_name() { + // Predicate indices can refer to an earlier projection: index 0 names the + // wrong file column, and usize::MAX is outside this file's schema entirely. + for index in [0, usize::MAX] { + let (logical_schema, batch) = schemas_and_batch(); + let inner: Arc<dyn PhysicalExprAdapterFactory> = + Arc::new(DefaultPhysicalExprAdapterFactory); + let column = Column::new("payload", index); + let static_filter: Arc<dyn PhysicalExpr> = Arc::new(BinaryExpr::new( + Arc::new(column.clone()), + Operator::Gt, + lit(15_i64), + )); + let original = inner + .create(Arc::clone(&logical_schema), batch.schema()) + .unwrap(); + let expected = original + .rewrite(Arc::clone(&static_filter)) + .unwrap() + .evaluate(&batch) + .unwrap() + .into_array(batch.num_rows()) + .unwrap(); + assert_eq!( + expected.as_any().downcast_ref::<BooleanArray>().unwrap(), + &BooleanArray::from(vec![false, true]) + ); + + let adapter = RuntimeFilterSchemaAdapterFactory::new(inner, vec![column]) + .create(logical_schema, batch.schema()) + .unwrap(); + let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new( + vec![Arc::new(Column::new("key", 0))], + lit(false), + )); + let combined: Arc<dyn PhysicalExpr> = Arc::new(BinaryExpr::new( + static_filter, + Operator::And, + dynamic_filter, + )); + let actual = adapter + .rewrite(combined) + .unwrap() + .evaluate(&batch) + .unwrap() + .into_array(batch.num_rows()) + .unwrap(); + // The converting payload disables only the dynamic filter. The static + // comparison must still select the second row using the named payload. + assert_eq!( + actual.as_any().downcast_ref::<BooleanArray>().unwrap(), + expected.as_any().downcast_ref::<BooleanArray>().unwrap(), + "stale column index {index}" + ); + } +} + +#[test] +fn unchanged_column_with_stale_index_keeps_live_filter() { + let (logical_schema, batch) = schemas_and_batch(); + let adapter = RuntimeFilterSchemaAdapterFactory::new( + Arc::new(DefaultPhysicalExprAdapterFactory), + vec![Column::new("key", usize::MAX)], + ) + .create(logical_schema, batch.schema()) + .unwrap(); + let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new( + vec![Arc::new(Column::new("key", 0))], + lit(true), + )); + let expr = Arc::clone(&dynamic_filter); + let adapted = adapter.rewrite(expr).unwrap(); + assert!(adapted.is::<DynamicFilterPhysicalExpr>()); + + dynamic_filter.update(lit(false)).unwrap(); + let actual = adapted + .evaluate(&batch) + .unwrap() + .into_array(batch.num_rows()) + .unwrap(); + assert_eq!( + actual.as_any().downcast_ref::<BooleanArray>().unwrap(), + &BooleanArray::from(vec![false, false]) + ); +} diff --git a/native/core/src/execution/operators/dynamic_filter/parquet_reader/schema_adapter/tests/resolution.rs b/native/core/src/execution/operators/dynamic_filter/parquet_reader/schema_adapter/tests/resolution.rs new file mode 100644 index 0000000000..21b7bcc631 --- /dev/null +++ b/native/core/src/execution/operators/dynamic_filter/parquet_reader/schema_adapter/tests/resolution.rs @@ -0,0 +1,133 @@ +// 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 super::*; +use crate::parquet::parquet_support::SparkParquetOptions; +use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory; +use datafusion::common::ScalarValue; +use datafusion_comet_spark_expr::EvalMode; +use parquet::arrow::PARQUET_FIELD_ID_META_KEY; +use std::collections::HashMap; + +fn spark_factory( + use_field_id: bool, + defaults: Option<HashMap<Column, ScalarValue>>, +) -> Arc<dyn PhysicalExprAdapterFactory> { + let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + options.use_field_id = use_field_id; + Arc::new(SparkPhysicalExprAdapterFactory::new(options, defaults)) +} + +fn assert_filter_result( + adapter: &Arc<dyn PhysicalExprAdapter>, + key: &str, + physical_schema: SchemaRef, + expected: bool, +) { + let predicate = Arc::new(DynamicFilterPhysicalExpr::new( + vec![Arc::new(Column::new(key, 0))], + lit(false), + )); + let batch = + RecordBatch::try_new(physical_schema, vec![Arc::new(Int32Array::from(vec![1]))]).unwrap(); + let actual = adapter + .rewrite(predicate) + .unwrap() + .evaluate(&batch) + .unwrap() + .into_array(1) + .unwrap(); + assert_eq!( + actual.as_any().downcast_ref::<BooleanArray>().unwrap(), + &BooleanArray::from(vec![expected]) + ); +} + +#[test] +fn unresolved_spark_column_disables_reader_filter() { + let schema = Arc::new(Schema::new(vec![Field::new("key", DataType::Int32, false)])); + let factory = spark_factory(false, None); + let column = Column::new("partition", 1); + let inner = factory + .create(Arc::clone(&schema), Arc::clone(&schema)) + .unwrap(); + // Spark's fallback can return an unknown name unchanged after the default + // adapter fails. A bare Column alone does not establish safe resolution. + let unresolved = inner.rewrite(Arc::new(column.clone())).unwrap(); + assert_eq!(unresolved.downcast_ref::<Column>(), Some(&column)); + let guarded = RuntimeFilterSchemaAdapterFactory::new(factory, vec![column]) + .create(Arc::clone(&schema), Arc::clone(&schema)) + .unwrap(); + assert_filter_result(&guarded, "key", schema, true); +} + +#[test] +fn remapped_physical_names_keep_reader_filter() { + for (physical_name, use_field_id) in [("KEY", false), ("stored_key", true)] { + let mut logical_field = Field::new("Key", DataType::Int32, false); + let mut physical_field = Field::new(physical_name, DataType::Int32, false); + if use_field_id { + let metadata = + HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "7".to_string())]); + logical_field = logical_field.with_metadata(metadata.clone()); + physical_field = physical_field.with_metadata(metadata); + } + let logical = Arc::new(Schema::new(vec![logical_field])); + let physical = Arc::new(Schema::new(vec![physical_field])); + let column = Column::new("Key", usize::MAX); + let factory = spark_factory(use_field_id, None); + let inner = factory + .create(Arc::clone(&logical), Arc::clone(&physical)) + .unwrap(); + let remapped = inner.rewrite(Arc::new(column.clone())).unwrap(); + assert_eq!( + remapped.downcast_ref::<Column>().unwrap().name(), + physical_name + ); + let guarded = RuntimeFilterSchemaAdapterFactory::new(factory, vec![column]) + .create(logical, Arc::clone(&physical)) + .unwrap(); + assert_filter_result(&guarded, "Key", physical, false); + } +} + +#[test] +fn missing_column_literals_keep_reader_filter() { + for default in [None, Some(ScalarValue::Int32(Some(7)))] { + let logical = Arc::new(Schema::new(vec![ + Field::new("key", DataType::Int32, false), + Field::new("missing", DataType::Int32, true), + ])); + let physical = Arc::new(Schema::new(vec![Field::new("key", DataType::Int32, false)])); + let column = Column::new("missing", 1); + let factory = spark_factory( + false, + default.map(|value| HashMap::from([(column.clone(), value)])), + ); + let inner = factory + .create(Arc::clone(&logical), Arc::clone(&physical)) + .unwrap(); + assert!(inner + .rewrite(Arc::new(column.clone())) + .unwrap() + .is::<Literal>()); + let guarded = RuntimeFilterSchemaAdapterFactory::new(factory, vec![column]) + .create(logical, Arc::clone(&physical)) + .unwrap(); + assert_filter_result(&guarded, "key", physical, false); + } +} diff --git a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala index 4c9b693198..2163ab05e2 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala @@ -25,6 +25,7 @@ import org.scalatest.Tag import org.apache.hadoop.fs.Path import org.apache.parquet.hadoop.ParquetFileReader import org.apache.parquet.hadoop.util.HadoopInputFile +import org.apache.spark.SparkException import org.apache.spark.sql.{CometTestBase, DataFrame, Row} import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation @@ -33,6 +34,7 @@ import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight} import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometFilterExec, CometHashJoinExec, CometNativeScanExec, CometSortMergeJoinExec, CometUnionExec} import org.apache.spark.sql.execution.{LocalTableScanExec, SparkPlan} import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AQEShuffleReadExec} +import org.apache.spark.sql.execution.datasources.SchemaColumnConvertNotSupportedException import org.apache.spark.sql.execution.exchange.ReusedExchangeExec import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.{ArrayType, IntegerType, MetadataBuilder, StructField, StructType} @@ -520,6 +522,62 @@ class CometJoinSuite extends CometTestBase { } } + test("join dynamic filter preserves Parquet schema conversion errors") { + withTempPath { probePath => + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", + SQLConf.LEAF_NODE_DEFAULT_PARALLELISM.key -> "1", + SQLConf.USE_V1_SOURCE_LIST.key -> "parquet", + SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") { + spark + .range(100, 104, 1, 1) + .selectExpr("CAST(id AS INT) AS probe_key", "id AS payload") + .write + .parquet(probePath.getCanonicalPath) + withTempView("dynamic_schema_probe") { + // INT64 -> INT32 is invalid even when the stored values fit. Keep the + // payload projected so discarding the nonmatching keys cannot hide it. + spark.read + .schema("probe_key INT, payload INT") + .parquet(probePath.getCanonicalPath) + .createOrReplaceTempView("dynamic_schema_probe") + withParquetTable(Seq(Tuple1(0)), "dynamic_schema_build") { + for ((comet, dynamicFilter) <- Seq((false, false), (true, false), (true, true))) { + withSQLConf( + CometConf.COMET_ENABLED.key -> comet.toString, + CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ENABLED.key -> dynamicFilter.toString) { + val df = sql( + "SELECT /*+ BROADCAST(b) */ p.probe_key, p.payload " + + "FROM dynamic_schema_probe p JOIN dynamic_schema_build b " + + "ON p.probe_key = b._1") + val plan = df.queryExecution.executedPlan + if (comet) { + val joins = collect(plan) { case join: CometBroadcastHashJoinExec => join } + assert(joins.size == 1, s"Expected one native broadcast hash join:\n$plan") + assert(joins.head.buildSide == BuildRight) + assert(joins.head.nativeOp.getHashJoin.getDynamicFilterEnabled == dynamicFilter) + val probes = collect(plan) { + case scan: CometNativeScanExec if scan.output.exists(_.name == "payload") => + scan + } + assert(probes.size == 1, s"Expected one native probe scan:\n$plan") + } + withClue(s"comet=$comet, dynamicFilter=$dynamicFilter: ") { + val error = intercept[SparkException](df.collect()) + val chain = causeChain(error) + assert( + chain.exists(_.isInstanceOf[SchemaColumnConvertNotSupportedException]), + s"Expected a Parquet schema conversion error, found $chain") + } + } + } + } + } + } + } + } + test("join dynamic filter preserves seeded rand probe order") { withTempPath { probePath => withSQLConf( --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
