fix: preserve Parquet conversion errors during join filtering

[pingzh]

Which issue does this PR close?

Part 2 of the five-PR plan for #5775, following the shared runtime-filter refactor in #5937. #5775 remains open for TopK fusion and reader pushdown.

Rationale for this change

Verification on Apache main at 6065705c16340c0be293212a71decfd9df4daae4 reproduced error suppression in existing join runtime filtering.

A Spark 4.1 broadcast join reads a Parquet INT64 payload as INT32 and has no matching probe keys. Vanilla Spark and Comet with runtime filtering disabled report SchemaColumnConvertNotSupportedException. Enabling the join runtime filter incorrectly lets the query succeed because reader pruning skips the incompatible data.

Native join tests also reproduce suppressed type-promotion errors in mixed-schema files and suppressed millisecond-to-microsecond timestamp overflow, including nested timestamps. These tests fail on the unmodified production code.

What changes are included in this PR?

• Wrap the existing schema-adapter factory for scans with an attached runtime filter. Check the projected columns and existing predicate dependencies separately for each file.
• Keep reader filtering for literals and direct column mappings whose rewritten names resolve in the physical file schema. Reject unresolved columns returned unchanged by Spark’s fallback adapter. For other adaptations, replace the reader's dynamic predicate with true and retain the original adapter, static predicates, and normal conversion-error timing. Decoded-batch runtime filtering still applies.
• Fall back to decoded-batch filtering when supplied file statistics can prune before the per-file adapter runs.
• Preserve static-predicate column names when checking conversions, use checked projection lookup, and add debug logs for unexpected attachment fallbacks.
• Add native and Spark regressions and document the conservative fallback, including allowed INT32-to-BIGINT promotion and struct subset projection. Track safe-adaptation support and a matching-schema fast path in perf: reduce Parquet runtime filter schema guard overhead #6123. No new configuration or metrics.

Empty files, statically excluded files, unprojected mismatches, and allowed conversions remain readable. Existing direct-schema reader-pruning tests still pass.

The separate missing-null-statistics investigation passed all 48 join configurations (96 executions). The fix therefore does not add the footer/statistics rewriting from the earlier TopK implementation.

How are these changes tested?

• Before the fix: four native regression tests failed on unchanged main; the readable controls and statistics investigation passed. The Spark regression failed specifically with Comet runtime filtering enabled after verifying that vanilla Spark and Comet filtering disabled raised the expected error.
• After the latest review follow-up: all 32 focused native tests passed, covering existing runtime-filter behavior, conversion regressions, partitioned scans, stale column indices, unresolved columns, case/field-ID mappings, missing/default literals, and planner metric export.
• The new real-Spark-adapter regression fails on c48d39f07: an unresolved column incorrectly keeps reader filtering enabled. Case/field-ID remapping and literal controls pass on both revisions.
• Mutation checks with the resolved-column fix: removing only the projected-partition exclusion makes projected_partition_keeps_runtime_reader_pruning fail its pruning assertion. Restoring it and removing only the static-predicate partition exclusion makes unprojected_partition_predicate_keeps_runtime_reader_pruning fail its pruning assertion. Both tests pass with both exclusions present.
• All 14 targeted Spark join dynamic-filter tests passed with Spark 4.1.3 and JDK 21, including the regression above. Verified that the packaged JNI library matches the rebuilt native library.
• Core crate Clippy passed for all targets with warnings denied; Cargo formatting, Maven Spotless/Scalastyle, Prettier, and git diff --check passed.
• Three independent sub-agents reviewed the latest guard, tests, and documentation, including the two additional reviewers requested before pushing. No actionable concerns remained.
• Relevant guard coverage comes from the native tests and CometJoinSuite, which explicitly enable join dynamic filtering. The default Spark SQL configuration leaves the feature disabled and does not exercise this guard. The broader run-spark-4.1-tests label has also been applied.

Results

TL;DR: Direct scalar pruning stayed effective. Safe promotions and nested narrowing read about 128× as much data and took 12–14× as long in these selective fixtures. The fully eligible wide selective case over many small files added 6.1 ms (+20.5%); the wide case with no row-group pruning showed +5.1%, with more timing variation.

Setup

• Compare Apache base 6065705c with c48d39f07 production changes using independently archived release native test binaries. These measurements predate the resolved-column follow-up and have not been rerun for it. Standard repository release profile: optimization level 3, thin LTO, one codegen unit. Rust 1.97.1, DataFusion 55.1.0, Arrow/Parquet 59.3.0.
• Linux x86-64, AMD EPYC 9V74, 32 CPUs available, about 126 GiB RAM. Local Parquet files, Snappy compression, dictionaries disabled, chunk statistics enabled, case-insensitive Spark schema adaptation. The benchmark measures native planning and execution on local files.
• One scan partition, batch size 8,192. Page-index pruning and row-filter pushdown disabled to isolate row-group pruning. The decoded-batch join filter remains enabled in the ON mode.
• Direct, promotion, and nested cases each have 16 files × 65,536 rows, with 8,192 rows per row group: 1,048,576 rows and 128 row groups. Promotion requests a BIGINT payload stored as INT32 with type promotion allowed. Nested narrowing requests two of eight INT32 struct children.
• Wide cases have 128 files × 2,048 rows × 64 INT32 columns, with 1,024 rows per row group: 262,144 rows and 256 row groups. Every column passes the guard. This exercises all eligibility checks for every file, without early failure.
• The build side contains key 0. Selective layouts have ordered unique probe keys, yielding one output row. The wide no-pruning layout alternates 0 and 1 inside every row group, retaining 50% of the probe rows (131,072 output rows) and preventing row-group pruning.
• Same files for both binaries. Execution order is base → head → head → base. Each process performs two warmups and seven measured executions per case/mode, alternating ON/OFF order. Reported times are medians of 14 measurements. Fresh plans are timed; file generation and session construction are excluded. OS and metadata caches are warm; measured metadata bytes are zero. Output row counts and attachment are asserted.

Timings and reader work

Data bytes are requested Parquet data ranges (scan_io_data_bytes, equal to bytes_scanned in these no-Bloom-filter fixtures), excluding metadata. Row groups read are total groups minus groups pruned; page and row filtering are disabled.

Case	Filter	Base ms	Head ms	Change	Data MiB base → head	Row groups read base → head
Direct scalars	OFF	16.68	16.19	-2.9%	8.007 → 8.007	128 → 128
Direct scalars	ON	1.05	1.06	+1.5%	0.063 → 0.063	1 → 1
Safe promotion	OFF	17.19	15.19	-11.7%	8.007 → 8.007	128 → 128
Safe promotion	ON	1.11	15.36	+1288.0%	0.063 → 8.007	1 → 128
Nested projection	OFF	21.34	15.11	-29.2%	12.011 → 12.011	128 → 128
Nested projection	ON	1.33	15.92	+1099.5%	0.094 → 12.011	1 → 128
Wide, selective	OFF	148.57	139.37	-6.2%	64.391 → 64.391	256 → 256
Wide, selective	ON	29.64	35.72	+20.5%	0.252 → 0.252	1 → 1
Wide, no RG pruning	OFF	151.08	154.51	+2.3%	63.439 → 63.439	256 → 256
Wide, no RG pruning	ON	186.27	195.83	+5.1%	63.439 → 63.439	256 → 256

OFF controls vary even though reader work is identical and the guard is inactive in that mode. Those differences include benchmark variability. In particular, promotion OFF was 12.59 ms in the first head process and 17.10 ms in the second. Small timing changes should be treated as indicative. The conversion cases' loss of pruning is deterministic. The wide selective overhead is consistent across passes: all head samples were 35.17–37.21 ms, while all base samples were 29.07–30.38 ms.

This deliberately conservative correctness guard has a cost for valid conversions and wide scans over many small files. Follow-up #6123 tracks accepting provably infallible adaptations and avoiding per-column rewrites for matching schemas. Nested leaf projection still works: the head reads only the two requested struct children; row-group pruning is disabled while nested column pruning remains effective.

Per-process median consistency (ms)

Case	Mode	First base	First head	Second head	Last base
Direct scalars	OFF	16.68	12.07	16.57	16.62
Direct scalars	ON	1.05	1.04	1.10	1.04
Safe promotion	OFF	17.38	12.59	17.10	17.00
Safe promotion	ON	1.12	13.14	17.75	1.09
Nested projection	OFF	21.56	14.99	17.97	20.86
Nested projection	ON	1.33	15.40	17.46	1.30
Wide, selective	OFF	141.16	137.32	139.40	152.18
Wide, selective	ON	29.44	35.73	35.59	29.81
Wide, no RG pruning	OFF	151.04	162.46	148.66	154.45
Wide, no RG pruning	ON	185.29	202.93	191.50	188.74

[sunchao]

Correctness

Prior state and problem

Reviewed e8dede5099c217f18d60fc94bfeafbf5ca1fce7d against 6065705c16340c0be293212a71decfd9df4daae4. The existing join runtime filter could discard a Parquet file or row group before decoding a projected payload column. A nonmatching join key could therefore hide an invalid Parquet-to-Spark schema conversion or a millisecond-to-microsecond timestamp overflow that the same scan would otherwise report.

The change preserves the existing conversion path and its error timing. It checks both projected columns and columns used by the static reader predicate, excludes partition columns that become literals, and decides independently for each file whether the original adapter performs only direct column remapping or produces a literal. For other adaptations, the wrapper replaces only the dynamic-filter expressions with true; static predicates, projection conversion and the decoded-batch runtime filter remain in place. Treating an unsuccessful eligibility rewrite as ineligible, rather than immediately propagating that error, is appropriate for empty files and files eliminated by the existing static predicate. Reader attachment and guard, per-file adapter.

Spark compatibility and boundaries

I checked the maintained Spark 3.5 and 4.0 sources. Their vectorized readers reject a Parquet INT64 payload requested as Spark INT32, even when its stored values fit, and use checked multiplication for timestamp milliseconds converted to microseconds. This supports the new regression's error-preservation contract; it is not a claim that every conversion or pruning configuration is identical across Spark versions. In particular, permitted type promotion differs by version, and the maintained 4.0 reader has its own runtime-filter path.

The new native tests cover mixed compatible/incompatible files, supplied file statistics, empty incompatible files, static exclusion, an unprojected incompatible column, allowed promotion, and both top-level and nested timestamp overflow. Existing passing controls cover null-check conjunctions with a remapped projection, retained residual filtering, limits and a seeded nondeterministic filter. The new Spark test also verifies the actual native broadcast join, build side, runtime-filter flag and native probe scan before checking the exception cause.

I checked the pre-adapter pruning boundary against the exact DataFusion 55.1.0 dependencies in Cargo.lock: FilePruner::try_new requires a supplied statistics object. Rejecting reader attachment for every statistics.is_some() therefore also covers objects containing only unknown statistics and partition-value folding. Without that object, the reader rewrites the predicate through the adapter before constructing its row-group/page/row pruning paths. I found no verified correctness issue in the changed path.

Validation and limits

The native CI job passed, and its log explicitly reports PASS for all five newly added native tests. The checkout was 74ad5cf3838e440b78862ee91713e3e5a59550e4, whose parents are precisely the reviewed base and head. Java lint jobs are also green. At the final 01:40 UTC check, 18 checks had succeeded, 13 were skipped, and six Spark integration/TPC validation jobs were running. The broader Spark SQL jobs remained skipped, so overall CI was not yet complete.

I did not run a local native build, Spark/JVM suite or performance benchmark. The PR author's reported Spark 4.1.3 execution is separate from the CI evidence above. Required maintained Spark 3.4 and 4.1 branches were unavailable for canonical-source comparison, so this review does not claim that comparison for those versions.

Performance

The change deliberately gives up reader-level pruning where it cannot establish that decoding is free of conversion failures. The normal direct-column case keeps reader pruning; a scan with any supplied file statistics uses batch filtering instead. The per-file eligibility check adds adapter rewrites for the deduplicated projected/static-filter column set, short-circuits at the first disallowed adaptation, and keeps the decision outside the row loop.

The allowlist is conservative: successful numeric promotion and pure nested structural narrowing can still produce a cast expression, so they also lose reader pruning. That is a source-demonstrated tradeoff, not a measured query-time regression. Could you add a focused microbenchmark comparing selective joins on direct scalar columns, safe promotion and nested projection, at the base and PR head, with runtime filtering enabled and disabled? Please record bytes and row groups read alongside query time so the cost of this conservative fallback is quantified before considering a wider allowlist. The runtime-filter option remains experimental and disabled by default.

Design

The design keeps conversion semantics in the existing Spark adapter and places the guard at the point where both logical and per-file physical schemas are available. This is simpler to reason about than duplicating conversion rules in the join or eagerly validating every file, either of which could diverge from static pruning and empty-file behavior.

The implementation separates two necessary cases: a scan-wide fallback for pruning that can happen before adapter creation, and a per-file fallback after the physical schema is known. It retains the original join as the authority for matching rows and keeps the existing limits, projections and nontrivial filter expressions as attachment boundaries. The tuning documentation now describes both fallbacks. There is no new configuration switch or change to exchange/FFI boundaries.

Abstraction & complexity

The small adapter-factory wrapper earns its place: it composes the existing factory, computes one per-file eligibility decision, and intercepts dynamic expressions without changing their shared producer or the residual batch filter. The error-conversion logic remains centralized. The dedicated schema-error and timestamp-error test modules keep the main join tests manageable.

I have no actionable abstraction or complexity finding. Any future refinement for infallible casts should reuse an explicit conversion-safety contract and retain these error-preservation regressions; expanding an ad hoc list of accepted expression shapes would require the same compatibility analysis.

[ajsquared]

Reviewed the exact head and complete current feedback. No independently confirmed P1 findings. The conversion guard preserves the original schema adapter and static predicates, with decoded-batch runtime filtering retained. Remote static review only; tests were not run, and CI and production were not inspected.

[andygrove]

I checked this out and ran it rather than reading only, so first the part that confirms your work. I reverted parquet_reader.rs to the merge base and removed parquet_reader/schema_adapter.rs while keeping your tests, and four of the five fail, with join_reader_schema_guard_preserves_readable_cases passing both ways. The suppressed-error bug is real and the regressions are not vacuous. I also confirmed against DataFusion 55.1.0 that the opener builds the adapter with table_schema.file_schema() as the logical schema, so the index space the eligibility check assumes is the right one.

One thing I would like covered before this goes in. read_columns.retain(|&index| index < file_column_count) is doing more than its one line suggests. read_columns holds table-schema indices, create resolves them against the logical file schema, and Schema::field panics rather than returning an error. I deleted that line locally and all 23 dynamic-filter tests still passed, while a partitioned scan panicked with index out of bounds: the len is 2 but the index is 2 inside arrow-schema. Partitioned tables are the normal Spark layout, so a regression there would be a panic in a native task rather than a lost pruning opportunity.

Could you add a partitioned case to schema_errors.rs? Three scenarios cover it: the partition column projected alongside the incompatible payload, the partition column projected with no file column needing a conversion, and the partition column referenced by the static reader predicate. All three pass on your head. I have a working version if that saves you time.

Related, would you consider logical_schema.fields().get(index) with None treated as ineligible? That turns any future index-space mismatch into a lost pruning opportunity instead of a panic, which is how the rest of the function treats the cases it cannot prove safe.

Building on @sunchao's benchmark request rather than duplicating it, it is worth making sure the measurement covers the wide fully eligible scan too. create runs once per file and does one inner.rewrite per read column, and all() short-circuits on the first ineligible column, so the case paying most for the check is the one where every column passes and no pruning is lost. Many columns over many small files is where that shows up.

Two smaller things. The expr_adapter_factory and post-pushdown downcast returns are the only bail-outs in try_attach_parquet_reader_filter without a log::debug!("Join dynamic filter reader pushdown skipped: ...") line. parquet_exec.rs:236 always sets the factory so neither should be reachable, which is the reason a log line would be worth having if that stops being true.

And for the static filter the code collects Columns, keeps only .index(), then rebuilds the name from that index in create. The default adapter resolves columns by name rather than index, so a predicate with a correct name and a stale index still works in DataFusion but would send the eligibility check to a different column. Carrying the collected Columns through would remove the assumption. I could not build a Comet plan that hits it, so this is about not depending on the invariant rather than a bug I can show you.

Separately, this touches a native operator and the Parquet scan path, so it needs a Spark SQL verdict before the merge queue rather than after. run-spark-4.1-tests needs to go on before this merges.

[pingzh]

hi @andygrove thanks for the comment, pushed another commit, could you please take another look 🙏

[andygrove]

Thanks for the rework. The Columns carried through, the checked .get(index), and the two debug logs all cover what I raised, and the three partition scenarios are the right three. I ran the suite rather than reading it, and all 29 dynamic-filter tests pass on c48d39f07.

One thing before this goes in, and it is the same class of gap as last time rather than a repeat of it. I deleted the if index < file_column_count guard at parquet_reader.rs:139, and separately the table_partition_cols() filter at parquet_reader.rs:146, and all 29 tests stayed green both times, including the three new partition ones. The refactor did remove the panic I found, so that part is genuinely fixed. What is left is that neither guard changes the decision, so nothing would notice if one stopped being correct.

The cause is in the allowlist rather than in the guards. SparkPhysicalExprAdapter::rewrite catches the default adapter's error and falls back to wrap_all_type_mismatches, which returns the expression untouched for a name it resolves in neither schema. An unguarded partition column therefore comes back as a bare Column and passes expr.is::<Column>(). The guard is reading "no adaptation needed" off a result that can also mean "could not resolve", which is the one reading the rest of the function is careful to avoid.

Would you consider requiring the rewritten column to resolve in the physical schema?

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,

I ran this locally. All 29 tests still pass, so it costs no pruning that you have coverage for. Deleting the projection guard then fails projected_partition_keeps_runtime_reader_pruning, and deleting the static-filter guard fails unprojected_partition_predicate_keeps_runtime_reader_pruning, so your three new tests start pinning the thing they were written for. It is also safe under case-insensitive matching, because the adapter remaps back to the original physical name before returning, so a logical Key against a file column KEY comes back as KEY@0 and still resolves.

For what it is worth, I could not turn the current looseness into a wrong answer. The case I tried, a non-nullable column missing from one file, panics in DataFusion's simplifier identically with the filter on and off, so it is pre-existing and not something this PR leaves open. The ask is about the contract rather than a bug I can show you.

On the tuning.md paragraph, "columns require conversions" reads as type conversions, and your benchmark shows it is broader than that. An allowed INT32 to BIGINT promotion rewrites to a Cast, and an eight-child struct projected down to two rewrites to CAST(c@0 AS Struct(...)), and both lose reader pruning even though neither can fail. Nested column pruning is very ordinary in Spark. Could the wording name those two cases, so someone tuning a query knows that projecting a subset of a struct's fields is enough to turn pruning off?

The benchmark itself answers what @sunchao and I asked for, and thank you for running it properly. Could you open a tracking issue for narrowing the cost and link it from the comment on allow_runtime_filter? Two things belong in it. Widening the allowlist to infallible adaptations, which is where the 12x to 14x on promotion and nested projection goes. And short-circuiting the per-column loop when the logical and physical schemas already match, which is where the +20.5% on the wide fully eligible scan goes, since create currently does one inner.rewrite per read column per file. #5775 covers TopK fusion and reader pushdown, so neither has a home today.

Last, please disregard what I said about needing run-spark-4.1-tests before this merges. I checked and I was wrong. apply_join_dynamic_filter returns early on !enabled, spark.comet.exec.join.dynamicFilter.enabled defaults to false, and nothing under dev/diffs/ sets it, so the Spark SQL suite cannot reach this code. Your CometJoinSuite test is the coverage that matters and it is already running in the PR tier. Sorry for sending you after a label you could not add.

[sunchao]

Re-reviewed c48d39f070732f6c5dd494fb6214e8cd3fef4270 against target base 6065705c16340c0be293212a71decfd9df4daae4, including the partition and index handling follow-up. I found no independently established P1/P2 introduced or materially worsened by this change. The added partition cases cover the projected-partition conversion error, safe reader pruning, and an unprojected partition predicate. The checked projection lookup and retained static column names address the reviewed index/name boundaries.

The guard preserves the original schema adapter and static predicates and keeps filtering decoded batches when reader pruning is disabled. I checked the pinned DataFusion pruning order and the configured Spark 3.4.3, 3.5.9, 4.0.4, and 4.1.3 Parquet conversion paths. The updated benchmark results document the conservative loss of pruning for adaptations that produce cast expressions, including promotion and nested narrowing; I did not repeat those benchmarks.

I also checked the latest review's unresolved-column concern. The Spark adapter fallback can return an unresolved name as a bare Column. Accepting that result can preserve the reader predicate already present at the target base, while actual predicate and projection adaptation still runs before pruning row groups. Current code excludes partition slots and names before this check. I did not establish an introduced or worsened P1/P2 from that contract or limited mutation coverage.

Local validation on the exact head passed the native library build, 29/29 focused native tests before and after the control, and 14/14 Spark 4.1.3 dynamic filter tests on JDK 17 with matching rebuilt and packaged JNI libraries. With only the reader attachment restored to the target base, five error regressions failed while 22 other focused tests passed; the reviewed source was then restored and verified clean. Two independent reviewers participated; the challenge pass was limited to source analysis and found no additional qualifying issue.

The initial PR CI run succeeded, including native and Spark 4.1 execution tests whose logs report the new regressions passing. Spark SQL run 35772505901 is still pending: its build and two test shards had reported success, and five Core/Hive shards were running at 19:57 UTC. That workflow does not enable the join dynamic filter, which is disabled by default, so the focused tests above provide validation with this guard enabled. Merge queue validation remains pending. The inspected CI logs use merge commit 0e912c96ab424a4d6845d030cedf1e004f8aed71, whose nine files changed by the PR match the reviewed head; the local runs used the head itself.

[sunchao]

Re-reviewed d04897e48357e511d4d85c6f38e03dec5b5cf51c against target base 6065705c16340c0be293212a71decfd9df4daae4, including the whole current diff and surrounding scan/join paths plus the delta from c48d39f. I found no P1/P2 issue.

The new check rejects unresolved bare columns while preserving the adapter's case matching and field ID remapping. The new resolution tests assert the guard decision, and the tuning text now names conservative fallback for safe promotion and struct narrowing and links #6123. I rechecked pruning from per-file statistics, projected columns, static predicate columns, partition values, conversion ordering, and runtime filter lifetime. A fresh independent source challenge found no additional candidate (2 reviewers total).

Local validation on this commit: the native library built and all 32 focused native tests passed; CometJoinSuite with the join dynamic filter name filter ran 14 tests on Spark 4.1.3 and JDK 17, all passing. I checked the suite XML and the packaged JNI checksum against the rebuilt library. The root Maven command exited 1 afterward on offline resolution of the unversioned exec-maven-plugin in the spark-integration packaging module; its jar contents check is bound to integration-test, so the completed focused suite remains valid.

At 21:06 UTC, GitHub checks associated with this head reported 22 successful, 12 skipped, and 9 unfinished, with no failed checks; Comet CI is still in progress. Other Spark profiles were not run locally. I did not run a performance benchmark for this commit.