Operator Tuning

Optimizing Joins

Spark often chooses SortMergeJoin over ShuffledHashJoin for stability reasons. If the build-side of a ShuffledHashJoin is very large then it could lead to OOM in Spark.

Vectorized query engines tend to perform better with ShuffledHashJoin, so for best performance it is often preferable to configure Comet to convert SortMergeJoin to ShuffledHashJoin. Comet does not yet provide spill-to-disk for ShuffledHashJoin so this could result in OOM. Also, SortMergeJoin may still be faster in some cases. It is best to test with both for your specific workloads.

To configure Comet to convert SortMergeJoin to ShuffledHashJoin, set spark.comet.exec.forceShuffledHashJoin=true. The conversion only happens when the build side is under a size limit, and a join whose build side has no statistics is left as a SortMergeJoin. The size is Spark’s planning estimate of the build side, or under AQE the materialized shuffle size of the build side. By default the limit is Spark’s own rule for choosing a ShuffledHashJoin: spark.sql.autoBroadcastJoinThreshold times the initial shuffle partition count (spark.sql.adaptive.coalescePartitions.initialPartitionNum when AQE and partition coalescing are both on and it is set, else spark.sql.shuffle.partitions). When broadcasts are disabled with a non-positive threshold, Spark’s default threshold of 10 MB is used instead. Set spark.comet.exec.forceShuffledHashJoin.maxBuildSize to a size in bytes to use a fixed limit, or to a non-positive value to convert every eligible join regardless of size.

Join Runtime Filters

Set spark.comet.exec.join.dynamicFilter.enabled=true to try experimental native hash join runtime filtering. It is disabled by default. Eligible joins are inner joins with one direct signed integer key (TINYINT, SMALLINT, INT, or BIGINT) and one native partition per input within each task. Both broadcast and shuffled hash joins support either Spark build side. Unsupported joins keep their existing execution path.

Once the build completes, its key domain filters probe batches before the hash probe. Eligible native Parquet readers also use the domain to prune row groups. Reader attachment can pass through direct-column IS NOT NULL checks, including conjunctions, and remaps columns when the scan itself projects the file schema. The original null checks and residual runtime filter remain in place. 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 with the setting disabled to distinguish reduced hash-probe work from reader I/O savings.

Early filtering in join chains

With spark.comet.exec.join.dynamicFilter.enabled=true, an eligible join can also use its completed build keys to filter decoded rows before an intermediate inner join. For example, when a final join accepts only customer 42, it can remove other customers before an earlier join looks up their orders. This reduces intermediate join work while the final join still checks every match and preserves duplicates.

Placement follows direct probe-side columns through ordinary inner joins already eligible for runtime filtering (a single signed-integer key), column-only projections, and direct column null checks above the first intermediate join. After crossing a join, the early consumer stays above inferred null checks so null keys do not prevent adaptive bypass. It stops at computed expressions, other predicates, limits, unsupported joins, and execution boundaries. These early filters operate on decoded batches; they do not introduce additional Parquet pruning or skip schema conversions. Each intermediate join retains its existing reader-filter checks.

Early filtering stops evaluating after two consecutive nonempty evaluated batches remove no rows, avoiding repeated membership checks on unselective inputs. That decision is local to one execution. The final join remains responsible for matching rows, including if later batches become more selective. The dynamic_filter_early_* metrics belong to the join supplying the filter; intermediate joins and projections retain their own input/output metrics.

Join runtime filters through residual predicates

Native join runtime filtering is enabled with spark.comet.exec.join.dynamicFilter.enabled. It can reject probe rows whose keys are absent from the completed build side and use eligible Parquet readers to prune row groups. By default, an intervening residual filter permits reader propagation only for direct-column IS NOT NULL checks combined with AND.

Set spark.comet.exec.join.dynamicFilter.allowDeterministicFilterPushdown=true to let the join filter pass through other deterministic residual predicates, such as quantity > 0 or a Spark-generated Bloom check. The original residual and the exact join remain in place. For example, when the build contains only item IDs 17 and 42, a row for item 99 can be rejected before evaluating its quantity. Nondeterministic predicates, limits, all standalone projections (including column-only projections), and execution boundaries continue to stop reader propagation. Spark can insert a standalone projection to drop residual-only columns, so the example requires keeping quantity in the output alongside item_id. Projection within the Parquet scan itself remains supported. Parquet schema conversion checks remain active. Reader attachment through column-only projections is tracked in #7113.

The separate setting lets applications keep the existing runtime-filter behavior while choosing whether to allow broader reader pruning. Spark can already skip residual evaluation through other pruning paths, but those paths are not present in every plan. This setting defaults to false because deterministic expressions can throw. Earlier pruning can skip division-by-zero or ANSI cast errors on rows that cannot join. Errors on retained rows still propagate. Enable it only when skipping such errors on eliminated rows is acceptable. Parquet row filtering during decoding also requires spark.comet.parquet.rowFilterPushdown.enabled; row-group statistics pruning does not require that setting.

Adaptive Partial Aggregation

Set spark.comet.exec.aggregate.skipPartial.enabled=true to let Comet bypass partial hash aggregation for high-cardinality grouping when it is not reducing the number of rows enough. This experimental optimization is disabled by default. It currently applies only to fused native shuffle-writer plans whose partial aggregates are grouping-only or single-argument COUNT. Low-cardinality inputs continue to aggregate normally. The SQL metric rows bypassing partial aggregation shows whether skipping occurred.

DataFusion makes the decision separately in each task. It starts checking after the first 100,000 input rows, and as soon as the number of groups divided by the number of input rows exceeds 0.8, it stops aggregating and sends the rest of the task’s rows to the shuffle as they are. It does not check again, so a task whose keys repeat after a mostly distinct start, such as several snapshot files of the same keys packed into one split, can shuffle many times more rows than it would with skipping disabled. Compare the shuffle write metrics with the setting enabled and disabled before enabling it for a workload.

Eligibility is conservative for the whole fused native plan: any unsupported partial accumulator, Spark PartialMerge, or mixed-mode aggregate disables skipping in that plan. Multi-argument COUNT and other accumulators are not admitted. Distribution-required grouping-only stages still fully deduplicate, and non-native-shuffle plans retain ordinary aggregation.

To experiment with the thresholds, also enable spark.comet.exec.respectDataFusionConfigs, a development and testing option that defaults to false. For example, the following SQL settings pass through the default threshold values, which you can adjust:

SET spark.comet.exec.aggregate.skipPartial.enabled=true;
SET spark.comet.exec.respectDataFusionConfigs=true;
SET spark.comet.datafusion.execution.skip_partial_aggregation_probe_rows_threshold=100000;
SET spark.comet.datafusion.execution.skip_partial_aggregation_probe_ratio_threshold=0.8;

A lower row threshold allows an earlier decision; a lower ratio threshold makes skipping more likely. These settings only tune eligible plans. They cannot enable skipping while spark.comet.exec.aggregate.skipPartial.enabled is false, or for unsupported accumulators and modes.

Local TopK Fusion

Set spark.comet.exec.topK.fusion.enabled=true to run an eligible local TopK in the same native execution as its Parquet scan. This experimental optimization is disabled by default. It currently supports a direct native Parquet scan ordered by one signed integer column (TINYINT, SMALLINT, INT, or BIGINT). Both sort directions and null orderings are supported. Other inputs use the existing TopK execution path.

Each scan partition keeps enough candidates for both LIMIT and OFFSET. With multiple partitions, Comet shuffles those candidates and performs the final TopK. With one partition, it reuses the local ordering without building a second heap. The final stage applies the offset and output projection.

Fusion reduces the work of passing scan batches between native execution blocks. Fusion alone still reads all input rows. TopK reader pruning is enabled separately. Fusion can also reduce overlap between scan decoding and TopK processing, so some workloads may run slower. Compare enabled and disabled runs with your data layout, payload width, limit, and partition count before enabling it. The CometTopKBenchmark microbenchmark covers these cases with ascending, descending, and random layouts.

TopK Reader Pruning

Set both spark.comet.exec.topK.fusion.enabled=true and spark.comet.exec.topK.dynamicFilter.enabled=true to pass the local TopK’s improving threshold to its Parquet reader. Both options are experimental and disabled by default. Eligibility is the same single signed integer key described above. Each task creates a fresh threshold; it is not shared across Spark partitions or exchanges, or retained for later executions.

Once the heap contains enough candidates for LIMIT + OFFSET, the reader can skip later row groups whose statistics prove that no row can improve those candidates. Existing Parquet page-index and decoder-filter options can also use the predicate. This option adds no separate filter over decoded scan batches. TopK continues to select the final candidates.

Reader attachment is conservative. A scan with a fetch limit, supplied file statistics, or a static predicate other than direct column IS NOT NULL checks keeps the existing execution path. For each file, schema adaptation disables pruning if it could hide a conversion error in a projected or filtered column. Missing null counts remain unknown, which can prevent pruning even when min/max statistics are present. These cases can still execute a fused TopK.

Reader pruning is most useful when small K values and the file order establish a strong threshold early. Descending or random layouts for an ascending query can prune few or no groups, while still paying the cost of attaching and checking the predicate. Wider rows can increase the benefit when groups are skipped. Compare pruning with fused in CometTopKBenchmark to measure the reader effect, and compare both with unfused to include the cost of fusion. Check the scan’s emitted rows, bytes_scanned, row_groups_pruned_dynamic_filter, and row_groups_pruned_statistics alongside elapsed time; attachment alone does not demonstrate a saving. Pruning when later files open uses the TopK threshold already available and increments row_groups_pruned_statistics. With one row group per file, the dynamic counter can stay zero despite substantial TopK pruning. The statistics counter also includes other predicates, so compare with filtering disabled to assess TopK savings. See TopK metrics.

Optimizing Sorting on Floating-Point Values

Comet normalizes NaN payloads and signed zeros in FLOAT and DOUBLE ordering keys, including floating-point values nested in arrays and structs, so ORDER BY, window ordering and range partitioning on them match Spark and stay native even with spark.comet.exec.strictFloatingPoint=true. Only the comparison key is normalized; returned values keep their original NaN representation and zero sign.

A key that nests values in an array or struct whose type can hold a null element or field stays native under ASC NULLS LAST and DESC NULLS FIRST too: Comet separates the key’s root null placement from its nested ordering (#6476). A RANGE window frame that has to find a row’s peers over such a key still falls back to Spark (#6477).

sort_array sorts array elements rather than ordering rows. It follows Spark’s floating-point ordering as well, so it also stays native with spark.comet.exec.strictFloatingPoint=true.

The default array_sort comparator also runs natively. It puts null array elements last while comparing nulls inside nested arrays and structs below non-null values. A custom array_sort comparator or an array with a non-default string collation continues through the JVM codegen dispatcher.