Skip to content

Push down global COUNT(*) to footer row counts in vortex-spark (#9655) - #9656

Open
AadhiKat wants to merge 1 commit into
vortex-data:developfrom
AadhiKat:spark-countstar-pushdown
Open

Push down global COUNT(*) to footer row counts in vortex-spark (#9655)#9656
AadhiKat wants to merge 1 commit into
vortex-data:developfrom
AadhiKat:spark-countstar-pushdown

Conversation

@AadhiKat

Copy link
Copy Markdown

Closes #9655.

Implements SupportsPushDownAggregates on VortexScanBuilder so a global COUNT(*) is
answered from file footer metadata instead of scanning data.

Scope is deliberately the narrow, always-correct case:

  • exactly one CountStar, no grouping expressions, and no pushed predicates (a pushed filter
    changes 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);
  • partial pushdown only (supportCompletePushDown stays false): the scan emits one partial
    count per file and Spark performs the final summation, per the
    SupportsPushDownAggregates column-order contract.

Mechanics: build() returns a VortexCountStarScan that plans one input partition per resolved
file (same directory expansion as regular scans). Each partition opens a single-file
DataSource on the executor and reads DataSource.rowCount(). Single-file sources open their
only file eagerly, so the count is RowCount.Exact — the reader checkStates on exactness so
any 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 why
this 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.

pruneColumns becomes a no-op after a successful aggregate pushdown, since Spark re-prunes to
the 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, full
scan 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.

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>
@AadhiKat
AadhiKat force-pushed the spark-countstar-pushdown branch from b044edf to 81e5d61 Compare August 27, 2026 11:03
@AadhiKat

AadhiKat commented Aug 27, 2026

Copy link
Copy Markdown
Author

Could a maintainer apply changelog/performance? The Validate Changelog Label check needs it and I don't have triage permissions. Everything else is green (Java suite included). Thanks!

Re the CodSpeed flag: the one "regressed" benchmark (mul_u32_nonnull_avx512, WallTime on a hosted runner) is measurement jitter — this PR only touches Java files under java/vortex-spark/, so it can't affect Rust kernels; a sibling benchmark "improved" +20% in the same run.

@codspeed-hq

codspeed-hq Bot commented Aug 27, 2026

Copy link
Copy Markdown

Merging this PR will regress 1 benchmark

⚠️ Unknown Walltime execution environment detected

Using the Walltime instrument on standard Hosted Runners will lead to inconsistent data.

For the most accurate results, we recommend using CodSpeed Macro Runners: bare-metal machines fine-tuned for performance measurement consistency.

⚡ 2 improved benchmarks
❌ 1 regressed benchmark
✅ 2127 untouched benchmarks
⏩ 106 skipped benchmarks1
🗄️ 4 archived benchmarks run2

Warning

Please fix the performance issues or acknowledge them on CodSpeed.

Performance Changes

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)

Open in CodSpeed

Footnotes

  1. 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.

  2. 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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[Spark] COUNT(*) scans all rows instead of using footer row counts

1 participant