Push down global COUNT(*) to footer row counts in vortex-spark (#9655) - #9656
Push down global COUNT(*) to footer row counts in vortex-spark (#9655)#9656AadhiKat wants to merge 1 commit into
Conversation
Implements SupportsPushDownAggregates on VortexScanBuilder: a global COUNT(*) (single CountStar, no grouping, no pushed predicates) is answered by per-file footer row counts instead of scanning data. Partial pushdown: one input partition per resolved file emits its exact footer count as a one-row columnar batch and Spark performs the final summation. Tested on Spark 3.5.9/Scala 2.12 and Spark 4.1.2/Scala 2.13. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XXCQ38wbvMxMyVuVXni3j8 Signed-off-by: Aadhithya Hari <aadhikat@gmail.com>
b044edf to
81e5d61
Compare
|
Could a maintainer apply Re the CodSpeed flag: the one "regressed" benchmark ( |
Merging this PR will regress 1 benchmark
|
| Mode | Benchmark | BASE |
HEAD |
Efficiency | |
|---|---|---|---|---|---|
| ❌ | WallTime | mul_u32_nonnull_avx512 |
5.6 µs | 6.3 µs | -11.66% |
| ⚡ | WallTime | arrow_checked_add_u32_avx2[16384] |
21.3 µs | 17.7 µs | +20.4% |
| ⚡ | Simulation | take[duplicates/repeated/primitive/nonnull/chunks=16/indices=1000] |
245.7 µs | 205.4 µs | +19.61% |
Tip
Investigate this regression by commenting @codspeedbot fix this regression on this PR, or directly use the CodSpeed MCP with your agent.
Comparing AadhiKat:spark-countstar-pushdown (81e5d61) with develop (59a056d)
Footnotes
-
106 benchmarks were skipped, so the baseline results were used instead. If they were deleted from the codebase, click here and archive them to remove them from the performance reports. ↩
-
4 benchmarks were run, but are now archived. If they were deleted in another branch, consider rebasing to remove them from the report. Instead if they were added back, click here to restore them. ↩
Closes #9655.
Implements
SupportsPushDownAggregatesonVortexScanBuilderso a globalCOUNT(*)isanswered from file footer metadata instead of scanning data.
Scope is deliberately the narrow, always-correct case:
CountStar, no grouping expressions, and no pushed predicates (a pushed filterchanges the matching row count, so footer totals would over-count; partition-column filters
never block the pushdown since they are applied to directory paths before planning);
supportCompletePushDownstays false): the scan emits one partialcount per file and Spark performs the final summation, per the
SupportsPushDownAggregatescolumn-order contract.Mechanics:
build()returns aVortexCountStarScanthat plans one input partition per resolvedfile (same directory expansion as regular scans). Each partition opens a single-file
DataSourceon the executor and readsDataSource.rowCount(). Single-file sources open theironly file eagerly, so the count is
RowCount.Exact— the readercheckStates on exactness soany future change to that invariant fails loudly rather than returning a wrong count. (The
multi-file
DataSource.rowCount()extrapolates across deferred children, which is exactly whythis is done per-file rather than once on the driver.) The count is emitted as a one-row
columnar batch, keeping the connector columnar-only.
pruneColumnsbecomes a no-op after a successful aggregate pushdown, since Spark re-prunes tothe aggregation output schema whose fields correspond to no table column. Declined pushdowns
leave the regular scan path untouched.
Measured on a 2M-row × 804-column table (10 files, 1.5GB) on GCS, Spark 3.5.9, fresh
DataFrame per iteration, median of 3:
COUNT(*)drops from 11,944ms (stock 0.76.0, fullscan through JNI) to 406ms (this patch, per-file footer reads) — 29× — matching the
parquet and lance connectors' metadata-count behavior. With a cached plan the count itself
executes in ~40ms; the 406ms includes per-query planning/schema inference.
Tests: end-to-end pushed count with plan assertion, filtered/grouped fallbacks with correctness
checks, and builder-level accept/decline coverage (COUNT(col), MIN, multiple aggregates,
group-by, pushed-predicate interaction). Run against both Spark 3.5 (Scala 2.12) and
Spark 4.1 (Scala 2.13) variants.
Future work (out of scope here):
COUNT(col)from null counts, MIN/MAX from zone maps(mirroring the DuckDB aggregate pushdown), and complete pushdown for single-partition plans.
Note
AI assistance disclosure: this change was developed with an agentic AI tool (Claude Code);
design, review, and verification by the author. All tests pass locally on both Spark variants.