andygrove opened a new issue, #6908: URL: https://github.com/apache/datafusion-comet/issues/6908
### What is the problem the feature request solves? A native filter copies every surviving row of every column it outputs before the next operator sees the batch: DataFusion's `FilterExec` runs `filter_record_batch` on each batch. Spark's whole-stage codegen fuses the filter into its consumer, so a rejected row is a `continue` in the aggregation loop and nothing is copied. On a narrow, tall input where the consumer reads every column, that copy dominates. I measured it while benchmarking #6180 (native `AtLeastNNonNulls`) with a release build on Spark 4.1 at `local[1]`: 4 double columns by 16M rows with 2% nulls and 2% NaN per column, running `df.na.drop().agg(count(*), sum(c0), sum(c1), sum(c2), sum(c3))`, which keeps about 85% of rows. Medians of 12 interleaved runs: | | wall | process CPU | |---|---|---| | Spark | 685 ms | 697 ms | | Comet, filter falls back to Spark (main) | 695 ms | 888 ms | | Comet, native filter (#6180) | 1203 ms | 1206 ms | The native filter step took 364 ms with all four columns in its output, against 72 ms for the same predicate when only the row count is needed (`count(*)`, where #6396 already prunes the filter's output columns). So about 290 ms goes to copying roughly 57M surviving values, about 5 ns each. On a 16-column table keeping 72% of rows, the same copy cost about 1 ns per value. I suspect the difference is that arrow's filter kernel switches from gathering by index to copying slices above 80% selectivity (`FILTER_SLICES_SELECTIVITY_THRESHOLD`), and with scattered nulls those slices are short, but I haven't verified that. None of this is specific to `AtLeastNNonNulls`. Any native filter followed by an operator that reads all of its columns pays the copy, and narrow tables with high selectivity are where it shows most. ### Describe the potential solution Two directions, neither evaluated yet. For a filter feeding a grouped partial aggregate, the predicate could be passed as the `opt_filter` argument of `GroupsAccumulator::update_batch` instead of materializing the filtered batch. The plain `Accumulator::update_batch` used by keyless aggregates has no such argument, so that case would need something else. Separately, it would be worth checking whether arrow's slice strategy is the right choice at high selectivity with scattered nulls, which would be an arrow-rs change. ### Additional context Found while measuring #6180 for #6093. With the filter falling back, the native scan overlaps the Spark filter on a second core at `local[1]`, which is why the fallback's wall time looks better than its CPU time. -- 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]
