ganeshashree opened a new pull request, #57395:
URL: https://github.com/apache/spark/pull/57395
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR:
https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g.,
'[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a
faster review.
7. If you want to add a new configuration, please read the guideline first
for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the
guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section
is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster
reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class
hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other
DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
When reading from an in-memory (cached) relation,
`DefaultCachedBatchSerializer`
(`sql/core/.../execution/columnar/InMemoryRelation.scala`) maps each
selected output
attribute to its ordinal in the cached schema. Both conversion methods
computed this
with an O(n*m) algorithm that, for each of the `m` selected attributes,
rebuilt the full
list of `n` cached-schema `ExprId`s and linear-scanned it with `indexOf`:
```scala
val columnIndices =
selectedAttributes.map(a => cacheAttributes.map(o =>
o.exprId).indexOf(a.exprId)).toArray
```
This PR builds a single ExprId -> ordinal map once (O(n)) and looks up each
selected
attribute in O(1), for overall O(n+m) and no per-attribute list allocation:
```scala
val cacheAttributeOrdinals =
cacheAttributes.iterator.map(_.exprId).zipWithIndex.toMap
val columnIndices =
selectedAttributes.map(a => cacheAttributeOrdinals.getOrElse(a.exprId,
-1)).toArray
```
The same change is applied to both:
- convertCachedBatchToColumnarBatch (vectorized path)
- convertCachedBatchToInternalRow (row path)
`getOrElse(exprId, -1)` preserves the prior indexOf semantics of returning
-1 when
an attribute is absent.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
The old implementation is O(n*m) in time and transient allocation: for each
of the `m`
selected attributes it rebuilds the full `n`-element list of cached
`ExprId`s and
linear-scans it. The map-based version is O(n+m). A microbenchmark isolating
the
`columnIndices` computation (all columns selected in reverse order so
`indexOf` cannot
short-circuit; 20k calls per case; asserts the old and new implementations
produce
identical index arrays before timing) shows the expected quadratic-vs-linear
divergence:
```
OpenJDK 64-Bit Server VM 17.0.15+6-Ubuntu-0ubuntu120.04 on Linux
5.4.0-1160-aws-fips
Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
columnIndices for 50 columns (20000 calls): Best Time(ms) Avg Time(ms)
Relative
--------------------------------------------------------------------------------------
old: rebuild list + indexOf per attr (O(n*m)) 223 228
1.0X
new: exprId->ordinal map (O(n+m)) 83 87
2.7X
columnIndices for 200 columns (20000 calls): Best Time(ms) Avg Time(ms)
Relative
--------------------------------------------------------------------------------------
old: rebuild list + indexOf per attr (O(n*m)) 2828 2850
1.0X
new: exprId->ordinal map (O(n+m)) 339 344
8.3X
columnIndices for 500 columns (20000 calls): Best Time(ms) Avg Time(ms)
Relative
--------------------------------------------------------------------------------------
old: rebuild list + indexOf per attr (O(n*m)) 18881 18883
1.0X
new: exprId->ordinal map (O(n+m)) 832 844
22.7X
columnIndices for 1000 columns (20000 calls): Best Time(ms) Avg Time(ms)
Relative
--------------------------------------------------------------------------------------
old: rebuild list + indexOf per attr (O(n*m)) 75310 76013
1.0X
new: exprId->ordinal map (O(n+m)) 2146 2148
35.1X
```
To be clear about scope: `columnIndices` is computed once per scan execution
on the
driver , so in absolute terms this saves microseconds per query for typical
tables and up to ~3.7 ms per query only for a pathologically wide
1000-column cached relation. This is primarily a code-quality cleanup,
removing a quadratic algorithm and per-attribute list reallocation in favor
of the
obvious map lookup, with the benchmark included to confirm the algorithmic
improvement
rather than to claim a meaningful end-to-end speedup..
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as
new features, bug fixes, or other behavior changes. Documentation-only updates
are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes
- provide the console output, description and/or an example to show the
behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to
the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No. This is a behavior-preserving internal optimization; output schema,
ordering, and
data types are unchanged.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some
test cases that check the changes thoroughly including negative and positive
cases if possible.
If it was tested in a way different from regular unit tests, please clarify
how you tested step by step, ideally copy and paste-able, so that other
reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why
it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions
for the consistent environment, and the instructions could accord to:
https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
- `build/sbt sql/compile` passes.
- Existing `CachedBatchSerializerSuite` and `InMemoryColumnarQuerySuite` (36
tests,
including column pruning and reordering cases that directly exercise this
mapping)
pass.
- A focused micro-benchmark was written to isolate the column-index
computation, using
real `AttributeReference`/`ExprId` objects with all columns selected in
reverse order
(so `indexOf` cannot short-circuit at position 0), 20k calls per case. It
asserted the
old and new implementations produce identical index arrays before timing,
confirming
behavior is preserved, and measured 2.7x (50 columns) to 35.1x (1000
columns) speedups
(see the table above). The benchmark was a temporary verification aid and
is not
included in this PR.
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this
patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling
Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
Generated-by: Claude Code (Opus 4.8)
--
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]
