peterxcli opened a new issue, #5775: URL: https://github.com/apache/datafusion-comet/issues/5775
### What is the problem the feature request solves? DataFusion's TopK dynamic filter can exclude input that cannot beat the current top K rows. For queries such as `SELECT x FROM t ORDER BY x LIMIT 16`, this can avoid reading later Parquet row groups after a useful threshold is established. Comet already constructs DataFusion `SortExec` with a fetch for TopK, but its current Spark execution path prevents reader pushdown: `CometTakeOrderedAndProjectExec` runs its child separately and passes Arrow batches into a new native `TopKInput` plan. The Parquet reader is outside that plan. Connecting a filter only below the existing TopK cannot recover scan I/O already performed. ### Describe the potential solution First make the per-partition TopK and an eligible native scan part of the same native execution, while preserving Spark's local TopK, shuffle, final TopK, projection, and offset behavior. Then connect DataFusion's live TopK predicate to that reader using the attachment introduced in #5699. Start with one direct signed-integer sort key and native Parquet input. - Preserve the local candidate count and final global result across one and multiple Spark partitions, including LIMIT/OFFSET and already ordered inputs. Keep threshold state local to the native execution; do not push it across Spark exchanges or JVM/Arrow boundaries. - Let DataFusion update the threshold as the heap fills and improves. Use fresh producer/reader state for each execution and preserve Spark metric ownership. - Check ascending/descending order, ties, null placement, empty input, fewer than K rows, offsets, and repeated execution. Keep unsupported expressions/types conservative until their Spark semantics are validated. - Demonstrate the actual Spark native plan contains both TopK and its Parquet reader, and that the reader receives updates. Compare filtering off/on for favorable, random, and unfavorable layouts and different K values, measuring row groups/rows pruned, bytes scanned, and elapsed query time. ### Additional context Source verification against Comet `4c14448c2799e519d2648357db4442132beb4651` and DataFusion 55.0.0: - [CometTakeOrderedAndProjectExec](https://github.com/apache/datafusion-comet/blob/4c14448c2799e519d2648357db4442132beb4651/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala#L124-L191) calls `child.executeColumnar()` and creates separate native executions around the shuffle. [getTopKNativePlan](https://github.com/apache/datafusion-comet/blob/4c14448c2799e519d2648357db4442132beb4651/spark/src/main/scala/org/apache/spark/sql/comet/CometExecUtils.scala#L130-L169) creates `Scan(source = TopKInput)` plus `Sort(fetch = limit)`. - The [native planner](https://github.com/apache/datafusion-comet/blob/4c14448c2799e519d2648357db4442132beb4651/native/core/src/execution/planner.rs#L1566-L1600) constructs `SortExec.with_fetch` but does not attach its filter. A normal Comet sort [does not set fetch](https://github.com/apache/datafusion-comet/blob/4c14448c2799e519d2648357db4442132beb4651/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala#L1177-L1193), so it is not an alternative reader-connected TopK path today. - DataFusion [creates the dynamic filter in with_fetch](https://github.com/apache/datafusion/blob/55.0.0/datafusion/physical-plan/src/sorts/sort.rs#L1061-L1089) and [offers it for pushdown](https://github.com/apache/datafusion/blob/55.0.0/datafusion/physical-plan/src/sorts/sort.rs#L1520-L1549). Comet needs to connect producer and consumer explicitly. Follow-up to #5699. The scan benefit requires the execution-boundary work above; enabling a DataFusion configuration flag alone is insufficient. A scratch native test on this Comet revision manually built `SortExec(fetch=16)` over Comet's Parquet reader and attached its dynamic predicate through the existing reader helper. With DataFusion 55.0.0, an ascending INT32 key, 40 row groups of 1,024 rows, row filtering enabled, and page-index pruning disabled, ascending TopK returned identical results with filtering off/on. Reported `bytes_scanned` fell from 164,640 to 4,116 and scan output rows from 40,960 to 1,024 (97.5% reductions). This demonstrates the reader benefit when producer and reader share a native plan; the current Spark `TopKInput` boundary was deliberately absent in the probe. It is not evidence that current Spark TopK queries already benefit, nor an elapsed-time benchmark. One scratch test covering MIN, MAX, and TopK passed. -- 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]
