Native Parquet scan ignores session-level datafusion.execution.parquet.* options #4990

Describe the bug

Comet's native Parquet scan ignores all session-level datafusion.execution.parquet.* options. get_options() in native/core/src/parquet/parquet_exec.rs constructs a fresh default:

let mut table_parquet_options = TableParquetOptions::new();
table_parquet_options.global.coerce_int96 = Some("us".to_string());
table_parquet_options.global.coerce_int96_tz = Some("UTC".to_string());

ParquetSource reads reader settings from table_parquet_options.global (datafusion-datasource-parquet-54.0.0/src/source.rs:465 for max_predicate_cache_size, and similarly for the other reader options), so anything set on the session config is discarded for the scan.

pushdown_filters and reorder_filters are the exceptions, and only by accident: try_pushdown_filters consults state.config_options() directly and ORs the two sources together (source.rs:739-740):

let config_pushdown_enabled = config.execution.parquet.pushdown_filters;
let table_pushdown_enabled = self.pushdown_filters();
let pushdown_filters = table_pushdown_enabled || config_pushdown_enabled;

Every other reader option, including max_predicate_cache_size, bloom_filter_on_read, binary_as_string, schema_force_view_types, and force_filter_selections, is silently inert.

This interacts with the spark.comet.datafusion.* passthrough in jni_api.rs, which translates such keys into DataFusion session options. That passthrough is itself gated behind spark.comet.exec.respectDataFusionConfigs (default false, CometConf.scala:839), which is intentional and documented as a testing option. The bug is that even with that flag enabled, parquet reader options still do not take effect, so the escape hatch appears to work while doing nothing.

Steps to reproduce

With spark.comet.exec.respectDataFusionConfigs=true, set spark.comet.datafusion.execution.parquet.max_predicate_cache_size to 0, 104857600, and 1073741824 in turn, with spark.comet.parquet.rowFilterPushdown.enabled=true, and run any query with a pushed filter.

The Predicate cache: records read from the cache and Predicate cache: rows physically read and decoded scan metrics are byte-for-byte identical across all three, confirming the setting never reaches the reader. On TPC-H SF100 Q6 all three reported 48,791,761,520 and 18,001,134,400 respectively.

A more direct check: set only spark.comet.datafusion.execution.parquet.pushdown_filters=true (leaving spark.comet.parquet.rowFilterPushdown.enabled unset). The scan still emits every row (600,037,902 for SF100 lineitem), showing the passthrough had no effect.

Expected behavior

Session-level parquet reader options should reach the scan, so that spark.comet.datafusion.execution.parquet.* behaves as its implementation intends when respectDataFusionConfigs is enabled.

Additional context

Found while investigating #4988.

A naive fix is not correct. Seeding the global options from the session config:

table_parquet_options.global = session_ctx
    .state()
    .config_options()
    .execution
    .parquet
    .clone();

makes the options reachable (verified: max_predicate_cache_size=0 then genuinely disables the cache and the metrics disappear), but it also regressed TPC-H SF100 Q6 with row-level pushdown enabled by ~51%, from 1.517 s to ~2.30 s (median of last 5 of 10 iterations). Reverting and re-measuring returned it to 1.517 s, so the attribution is solid.

The cause is not understood. It is not pushdown_filters, which was already effectively enabled through the OR shown above, and it is not reorder_filters, which measures as a no-op on this query (2.302 s vs 2.303 s). Something else among the seeded defaults is responsible, and it should be identified before any fix lands, since whatever it is may also be costing performance in configurations where a user has legitimately set these options.

Measured on DataFusion 54.0.0, arrow/parquet 58.3.0, Spark 3.5.8, TPC-H SF100 local Parquet.