andygrove commented on PR #6067:
URL:
https://github.com/apache/datafusion-comet/pull/6067#issuecomment-5778786789
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 `Column`s, 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 `Column`s
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
