andygrove commented on code in PR #5107:
URL: https://github.com/apache/datafusion-comet/pull/5107#discussion_r3684688656
##########
native/core/src/execution/jni_api.rs:
##########
@@ -590,11 +590,22 @@ fn prepare_datafusion_session_context(
// Translate the Comet-namespaced row-level pushdown flag into the
equivalent
// DataFusion session options. `pushdown_filters` enables the parquet
reader's
// RowFilter evaluation during decode (late materialization);
`reorder_filters`
- // is only meaningful when pushdown_filters is on, so they move together.
+ // is only meaningful when pushdown_filters is on, so they move together.
Skip a
+ // key the `spark.comet.datafusion.*` testing escape hatch above already
set
+ // explicitly, so an explicit override isn't silently forced back to
`true`.
if spark_config.get_bool(COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED) {
- session_config = session_config
- .set_str("datafusion.execution.parquet.pushdown_filters", "true")
- .set_str("datafusion.execution.parquet.reorder_filters", "true");
+ const PUSHDOWN_FILTERS_KEY: &str =
+ "spark.comet.datafusion.execution.parquet.pushdown_filters";
+ const REORDER_FILTERS_KEY: &str =
+ "spark.comet.datafusion.execution.parquet.reorder_filters";
+ if !spark_config.contains_key(PUSHDOWN_FILTERS_KEY) {
Review Comment:
Nice catch on the precedence bug. Could this be handled by ordering instead?
If the `rowFilterPushdown.enabled` translation moves above the
`SPARK_COMET_DF_PREFIX` pass-through loop, then an explicitly-set
`spark.comet.datafusion.execution.parquet.pushdown_filters` naturally
overwrites the derived value, and you get the same behavior without the two
hardcoded key constants or the `contains_key` checks.
It would also cover any future Comet flag that derives a DataFusion setting,
rather than needing a new branch per key. As a bonus it removes the
`reorder_filters` branch, which currently has no test.
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -220,8 +225,30 @@ fn get_options(
allow_timestamp_ltz_to_ntz: bool,
object_store_url: &ObjectStoreUrl,
encryption_enabled: bool,
+ session_parquet_options: &ParquetOptions,
) -> (TableParquetOptions, SparkParquetOptions) {
let mut table_parquet_options = TableParquetOptions::new();
+
+ // Reader options that reach `ParquetSource`'s actual per-file execution
path
+ // (`ParquetMorselizer`, built from `table_parquet_options.global`) rather
than only
+ // `ParquetFormat::infer_schema`'s schema-inference path, which Comet
never calls (Comet
+ // always supplies its own schema, translated from Spark's catalog).
Seeded from the
+ // session so `spark.comet.datafusion.execution.parquet.*` (behind
+ // `spark.comet.exec.respectDataFusionConfigs`) and `spark.comet.parquet.
+ // rowFilterPushdown.enabled` actually take effect on the native scan
(#4990); a fresh
+ // `TableParquetOptions::new()` ignored the session entirely.
+ table_parquet_options.global.pushdown_filters =
session_parquet_options.pushdown_filters;
Review Comment:
The per-field verification behind this list is genuinely useful, thanks for
spelling it out in the description. My concern is what happens six months from
now. When DataFusion adds a reader option, nothing here fails, and the symptom
is the same silent inertness that #4990 was about.
Would `table_parquet_options.global = session_parquet_options.clone()` work,
keeping the `coerce_int96` / `coerce_int96_tz` assignments right after it? From
what I can see the extra fields are all harmless. Everything from
`data_pagesize_limit` down in `ParquetOptions` is writer-side and this scan
never writes. The four `infer_schema` fields you called out are no-ops for the
reason you gave. And `global.metadata_size_hint` is not read by `ParquetSource`
at all, since it keeps its own `metadata_size_hint` field that the
`with_metadata_size_hint(512 * 1024)` call above sets.
That way the Spark-compat overrides still win because they come last, and
the allowlist cannot drift.
##########
spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala:
##########
@@ -829,6 +829,78 @@ class CometNativeReaderSuite extends CometTestBase with
AdaptiveSparkPlanHelper
}
}
+ test("row-level pushdown reaches native scan when rowFilterPushdown.enabled
is set") {
Review Comment:
Both new tests are good and they cover the #4990 regression directly.
The Rust unit test proves `get_options` copies the fields, though, which is
a narrower claim than a user-set option actually changing what the reader does.
Five of the seven newly-plumbed options have no coverage at either level.
It might be worth one more test that flips a non-pushdown option through the
escape hatch and asserts observable behavior.
`spark.comet.datafusion.execution.parquet.pruning=false` against the
multi-row-group setup from `"row-group statistics pruning fires for native
Parquet scan dataFilters"` would work, asserting `row_groups_pruned_statistics
== 0`. That pins the plumbing at the level users care about, and it would have
failed before this PR.
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -220,8 +225,30 @@ fn get_options(
allow_timestamp_ltz_to_ntz: bool,
object_store_url: &ObjectStoreUrl,
encryption_enabled: bool,
+ session_parquet_options: &ParquetOptions,
) -> (TableParquetOptions, SparkParquetOptions) {
let mut table_parquet_options = TableParquetOptions::new();
+
+ // Reader options that reach `ParquetSource`'s actual per-file execution
path
+ // (`ParquetMorselizer`, built from `table_parquet_options.global`) rather
than only
+ // `ParquetFormat::infer_schema`'s schema-inference path, which Comet
never calls (Comet
+ // always supplies its own schema, translated from Spark's catalog).
Seeded from the
+ // session so `spark.comet.datafusion.execution.parquet.*` (behind
+ // `spark.comet.exec.respectDataFusionConfigs`) and `spark.comet.parquet.
+ // rowFilterPushdown.enabled` actually take effect on the native scan
(#4990); a fresh
+ // `TableParquetOptions::new()` ignored the session entirely.
+ table_parquet_options.global.pushdown_filters =
session_parquet_options.pushdown_filters;
+ table_parquet_options.global.reorder_filters =
session_parquet_options.reorder_filters;
+ table_parquet_options.global.force_filter_selections =
+ session_parquet_options.force_filter_selections;
+ table_parquet_options.global.bloom_filter_on_read =
+ session_parquet_options.bloom_filter_on_read;
+ table_parquet_options.global.max_predicate_cache_size =
+ session_parquet_options.max_predicate_cache_size;
+ table_parquet_options.global.enable_page_index =
session_parquet_options.enable_page_index;
Review Comment:
Worth a word in the comment about this one. `EagerPageIndexReaderFactory`
forces `PageIndexPolicy::Optional` on every fetch and ignores the requested
policy, so a session `enable_page_index=false` stops the page index from being
used for pruning but does not stop it from being read.
Someone setting this to reduce metadata I/O would not get that. The comment
above says these fields reach the real execution path, which is accurate but
reads as a stronger guarantee than this particular field has.
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -76,6 +76,10 @@ pub(crate) fn init_datasource_exec(
use_field_id: bool,
ignore_missing_field_id: bool,
) -> Result<Arc<DataSourceExec>, ExecutionError> {
+ // Computed once and reused below for `try_pushdown_filters`, since
`SessionContext::state()`
+ // clones the whole `SessionState` (function registries, optimizer rules,
etc.), not just the
+ // config.
+ let state = session_ctx.state();
Review Comment:
Before this change `state()` was only reached when there were data filters.
It is now on every scan, including filter-free ones.
`SessionState::clone` copies eight owned `HashMap`s, and `scalar_functions`
holds every DataFusion builtin plus the roughly 250 UDFs registered in
`prepare_datafusion_session_context`. Since `init_datasource_exec` runs once
per native plan creation, that is once per Spark task.
Would `session_ctx.copied_config()` do here? It clones only the
`SessionConfig`, and `.options()` gives you the `&ConfigOptions` that both
`get_options` and `try_pushdown_filters` need.
##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -152,7 +152,10 @@ object CometConf extends ShimCometConf {
"statistics, page index, bloom filters) is independent of this flag
and runs " +
"whenever Spark's spark.sql.parquet.filterPushdown is enabled.
Disabling this " +
"flag still lets format-level pruning work; the per-row eval falls
back to " +
- "the CometFilter operator above the scan.")
+ "the CometFilter operator above the scan. This flag's derived
pushdown_filters " +
+ "/ reorder_filters settings can be overridden by explicitly setting
" +
+ "spark.comet.datafusion.execution.parquet.pushdown_filters /
reorder_filters " +
+ "when spark.comet.exec.respectDataFusionConfigs is enabled.")
Review Comment:
`COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED` is `CATEGORY_PARQUET`, so this
text ends up in the published config reference, while
`spark.comet.exec.respectDataFusionConfigs` is `CATEGORY_TESTING` and is
described in its own doc as a development and testing option.
Does pointing users at it from a user-facing page read as an endorsement we
do not want? The precedence behavior is worth writing down, so maybe the
contributor guide is a better home for it.
--
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]
