JeonDaehong opened a new issue, #18026:
URL: https://github.com/apache/iceberg/issues/18026
### Feature Request / Improvement
### Summary
`ColumnarBatchUtil.buildRowIdMapping` and `buildIsDeleted` call
`PositionDeleteIndex.isDeleted(pos)` once for every row in a batch. Positions
within a batch are a contiguous ascending range and the DV-backed index is a
Roaring bitmap, so the same information can be obtained with a single range
traversal instead of `batchSize` independent probes.
On a V3 table read with a narrow projection, this loop accounts for **43–58%
of scan CPU**, depending on delete density — a figure I have since reproduced
on three CPU microarchitectures (Zen 3, Sapphire Rapids, Graviton3), on local
disk and on S3, with a warm and a dropped page cache, and at task parallelism 1
through 16. It moves by at most a few points across all of those, including
with a heavy aggregation in the same job. Two things do move it: on a small
distributed cluster it comes out at **42%** (the top of the range is a
single-JVM figure — executor startup and serialization enlarge the
denominator), and on a table split into many small files it drops by about **11
points** (per-file CPU — footer parsing and one DV load per file — also
enlarges the denominator). I implemented the change against
`apache-iceberg-1.11.0` and measured it end to end across delete densities from
0.5% to 50%: the delete-check CPU drops by **2.6x–9.3x** for
`buildRowIdMapping` and **1
4.9x–18.9x** for `buildIsDeleted`, and the scan subtree as a whole drops by
**12–45%**. Tables with equality deletes keep the existing loop and show no
regression.
There is also something users can do today, without waiting for this:
**sorting the table by the column the deletes target makes the delete check
2.8x cheaper on its own**, because the Roaring bitmap switches from array to
run containers. That only works when the deletes concentrate on relatively few
distinct key values — I measured it fading to 1.1x and then to nothing as the
sort key's cardinality rises. The two remedies overlap but do not replace each
other: sorting alone 2.8x, this patch alone 7.8x, both together 12.4x.
How much of that reaches end-to-end query time depends on how much work the
rest of the scan does. The threshold is not a column count — it is the share
the delete check holds in scan CPU. Above roughly 20% the change is worth
**13–19% of end-to-end query time**; below it the difference falls inside my
measurement noise, even though the delete-check CPU is still reduced 3.0x–6.7x
there. The change removes the same absolute CPU either way.
As a unit that transfers to other schemas: **one delete check costs about as
much as decoding 1.6–3.2 fixed-width integer columns**, while a single 32-char
string column costs 8–10 integer columns. On my table a projection of ten
integer columns still shows the effect and a projection of three md5 strings
does not — so a column count is the wrong thing to quote. One caveat on reading
any CPU figure below as latency: across 24 configurations only about half of a
profiler-measured CPU saving arrives as wall clock (regression slope 0.53).
`RoaringBitmap` already provides the range APIs needed (`forEachInRange`,
`forAllInRange`, `rangeCardinality`), and they are present in the version
Iceberg pins (`1.6.14`), including the shaded copy in `iceberg-spark-runtime`.
I have a working branch with the change, updated tests, and the measurements
below. Happy to open a PR if the direction sounds reasonable.
### Where
`spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchUtil.java`
(same shape in v3.5/v4.1/v4.2):
```java
// :57 buildRowIdMapping
PositionDeleteIndex deletedPositions = deletes.deletedRowPositions(); //
:66
for (int rowId = 0; rowId < batchSize; rowId++) { //
:72
long pos = rowStartPosInBatch + rowId;
...
if (isDeleted(pos, row, deletedPositions, eqDeleteFilter)) { ... }
}
// :110 buildIsDeleted — same loop shape at :125
// :137
private static boolean isDeleted(...) {
if (deletedPositions != null && deletedPositions.isDeleted(pos)) { //
:143
```
### Why it costs what it does
`isDeleted(pos)` resolves to `BitmapPositionDeleteIndex.isDeleted` →
`RoaringPositionBitmap.contains` → `RoaringBitmap.contains`, and each call
independently:
1. extracts the high/low key and bounds-checks the `RoaringBitmap[]`
(`RoaringPositionBitmap.contains`),
2. binary searches the top-level container array (`RoaringArray.getIndex` →
`binarySearch`),
3. searches within the container.
None of that is amortized across the batch even though every call lands in
the same one or two containers.
The cost also depends heavily on which container the deletes materialized
into, which is a function of delete density — `ArrayContainer` below 4096
entries per 65536-position chunk, `BitmapContainer` above it:
| chunk density | cardinality | container | per-row `contains` |
|---|---|---|---|
| 0.5% | 327 | array | 13.0 ns |
| 5% | 3,276 | array | 16.0 ns |
| 12% | 7,864 | bitmap | 4.0 ns |
| contiguous 30% | 19,661 | run | 4.6 ns |
So the sparse-delete case that CDC workloads spend most of their time in is
also the most expensive per row, because it lands on array containers and pays
a binary search per probe.
<details>
<summary><b>Evidence — cost as a function of delete density</b> (26
lines)</summary>
Spark 4.0.4, Iceberg 1.11.0, JDK 17, `format-version=3`, MoR. 8M rows across
4 files (~30 Roaring chunks per file). async-profiler, `ctimer`. Scan
materialized through the `noop` sink; the vectorized reader was confirmed
active in every run. Each configuration profiled **3 times in separate JVMs**;
the run-to-run spread of this measurement is 9.7% median (see *Caveats*), so
ranges are reported and differences inside that band are not claimed.
`buildRowIdMapping` probes `batchSize` times per batch regardless of
density, and the total scanned row count is fixed, so the sample count is
directly proportional to per-probe cost.
| delete density | dominant container | delete-check CPU samples | share of
scan subtree |
|---|---|---|---|
| 0% (control) | — | **0** | 0.00% |
| 0.5% | array | 599 (576–633) | 44.2% |
| 5% | array | 746 (700–798) | 47.5% |
| **6.1%** | array | **872 (838–909)** | **53.2%** |
| 6.25% | mixed | 654 (617–702) | 46.2% |
| 7.0% | bitmap | 394 (381–406) | 33.6% |
| 12% | bitmap | 422 (410–447) | 36.3% |
| 50% | bitmap | 496 (454–529) | 39.6% |
The control table with no deletes shows exactly zero samples attributed to
the delete path, so the attribution has no false positives.
Two things worth noting:
- The worst case is **just below the array→bitmap boundary** (~6.1% deletes
per chunk), not at high delete rates. Deleting *more* rows can make the per-row
check cheaper.
- With a wide projection (20 columns) the same table shows 3.75%, because
the check is per row regardless of how many columns are read. **The 40–53%
figures are specific to narrow projections.** See *Evidence — projection width*
below for the patched numbers at 20 columns.
Wall clock, 1 column, median of 12 iterations: a table with 0.5% deletes
reads **13.0% slower** than the same table with no deletes at all — despite
having 0.5% fewer rows to emit.
</details>
<details>
<summary><b>Evidence — with the change applied</b> (51 lines)</summary>
I cloned `apache-iceberg-1.11.0` (commit `6976e02`), implemented the change,
and re-measured the **same tables** with only the jar swapped. Both jars were
built from the same tree — the baseline is the same source with the patch
stashed — so the only difference between them is the diff.
Correctness was checked first: `count(*)`, `sum(id)` and `min/max(id)` over
7 tables, identical for both jars. `sum(id)` is sensitive to *which* rows
survive, not just how many.
| configuration | container | current | patched | reduction | share of scan
subtree | scan subtree |
|---|---|---|---|---|---|---|
| 0.5% deletes | array | 674 (588–748) | **79 (67–97)** | **8.6x** | 43.2% →
7.6% | −33.6% |
| **6.1% deletes** | array | **873 (704–959)** | **94 (68–117)** | **9.3x**
| **49.4% → 9.6%** | **−44.8%** |
| 7.0% deletes | bitmap | 420 (378–474) | 113 (109–115) | 3.7x | 33.4% →
11.5% | −22.0% |
| 0.5%, unclustered | array | 621 (568–683) | 73 (64–78) | 8.5x | 44.1% →
7.4% | −29.9% |
| 0.5%, clustered | run | 185 (174–196) | 61 (43–79) | 3.1x | 19.1% → 7.0% |
−12.4% |
3 repetitions per arm, arms interleaved (baseline r1, patched r1, baseline
r2, …) so machine drift is shared rather than attributed to the jar. The ranges
of the two arms do not overlap in any configuration; comparing the best
baseline against the worst patched run still leaves 2.2x–7.3x.
The gain tracks the baseline cost: it is largest on array containers, which
is exactly where the removed work — a binary search per row — was most
expensive.
Those densities all sit near the container boundary, so I later filled in
the high-delete range as well (1 column, 4 repetitions, both densities on
bitmap containers so container type is held fixed):
| delete density | container | current | patched | reduction |
|---|---|---|---|---|
| 8.0% | bitmap | 499 (489–552) | 142 (137–146) | **3.5x** |
| 50.0% | bitmap | 584 (556–612) | 225 (200–270) | **2.6x** |
So the change helps across the whole density range, and **2.6x is the
floor**, reached where deletes are so dense that there are no gaps left to skip
— at 50% deletes the average run of live rows is one row, so the bulk range API
loses its main advantage and what remains is the removal of the per-row lookup
and branch. The largest gain is at 6.1%, which is also where the current code
is worst.
Hot frames before and after (0.5% deletes, 1 column, sample counts):
```
current patched
RoaringBitmap.contains 59 RoaringBitmap.forEachInRange
13
Util.unsignedBinarySearch 49 RoaringBitmap.forAllInRange
13
Util.hybridUnsignedBinarySearch 45 ArrayContainer.forAllInRange
11
ArrayContainer.contains 34
IntConsumerRelativeRangeAdapter 11
RoaringArray.binarySearch 17
Util.hybridUnsignedBinarySearch 1
RoaringArray.getContainerIndex 17 Util.unsignedBinarySearch
1
```
Binary-search samples: **94 → 2**.
Wall clock, 1 column, median over 30 iterations × 3 runs:
| configuration | current | patched |
|---|---|---|
| 0.5% deletes | 0.207 s | 0.180 s (−12.8%) |
| **6.1% deletes** | 0.221 s | **0.153 s (−30.8%)** |
| 7.0% deletes | 0.171 s | 0.150 s (−12.8%) |
These scans are short (0.15–0.22 s) so wall clock is noisier than the sample
counts; I treat it as a direction check rather than the primary number.
</details>
<details>
<summary><b>Evidence — projection width</b> (60 lines)</summary>
The numbers above are all `select` of a single column. Because the delete
check runs once per row regardless of projection width, widening the projection
leaves the delete-check cost alone and inflates everything around it. I re-ran
the same three tables projecting 20 columns, 6 repetitions per arm, with the
arm order flipped halfway (rounds 1–3 baseline first, rounds 4–6 patched first):
| configuration | delete-check share, 1 col | delete-check share, 20 cols |
reduction, 1 col | reduction, 20 cols |
|---|---|---|---|---|
| 0.5% deletes | 43.2% | **3.2%** | 8.6x | **6.8x** |
| **6.1% deletes** | 49.4% | **4.8%** | 9.3x | **6.3x** |
| 7.0% deletes | 33.4% | 2.7% | 3.7x | 3.5x |
The arm ranges are disjoint in all three configurations, so the reduction
itself still holds at 20 columns. The mechanism shows up directly in the
absolute sample counts: widening the projection 20x leaves the baseline
delete-check samples essentially unchanged (674→625, 873→902, 420→516) while
the scan subtree grows 1,558→19,332 — the share falls because the denominator
grew, not because the check got cheaper.
I then filled in the widths between, 6 repetitions per arm at each (108
profiles). End-to-end wall clock, paired within each round (negative = patched
faster; **bold** = outside the 9.7% noise band with all 6 pairs agreeing in
sign, sign test p = 0.03):
| projected columns | 0.5% deletes | 6.1% deletes | 7.0% deletes |
|---|---|---|---|
| 1 | **−19.5%** | **−18.8%** | **−12.8%** |
| 3 | **−18.5%** | **−13.5%** | −15.6% (5/6, p=0.22) |
| 5 | **−13.2%** | **−15.2%** | −6.7% |
| 10 | −2.3% | −6.7% | +4.7% |
| 20 | −8.4% | +3.5% | +4.9% |
So the crossover is between 5 and 10 columns on this table. But the useful
statement is not a column count — it is what those columns cost to decode. Scan
samples added per extra column, at 6.1% deletes:
| range | columns added | scan samples per column |
|---|---|---|
| 1 → 3 | 2 ints | 280 |
| 3 → 5 | 2 ints | 100 |
| **5 → 10** | 2 doubles + **3 md5 strings** | **1,958** |
| 10 → 20 | mixed | 660 |
The first five columns of this table are integers; the first string column
is the eighth — so column count and decoding cost are completely confounded in
the table above.
To separate them I ran a third experiment that puts the two in opposition:
ten integer columns (many columns, cheap) against one and three md5 string
columns (few columns, expensive), 6 repetitions per arm across the same three
densities (108 profiles). Delete-check share of scan CPU:
| projection | columns | 0.5% deletes | 6.1% deletes | 7.0% deletes |
|---|---|---|---|---|
| 1 int (`id`) | 1 | 43.2% | 49.4% | 33.4% |
| 1 md5 string | 1 | 17.4% | 22.2% | 12.1% |
| 3 ints | 3 | 35.5% | 42.8% | 27.6% |
| 3 md5 strings | 3 | 7.5% | 10.6% | 5.5% |
| **10 ints** | 10 | **15.9%** | **21.6%** | **12.8%** |
| 10 mixed (3 strings) | 10 | 5.9% | 7.7% | 4.3% |
Ten integer columns hold a **larger** share than three string columns in all
three densities — the opposite of what a column-count model predicts, and a
single integer column holds twice the share of a single string column. Solving
for per-column decode cost across these configurations gives 288–299 samples
for an integer column against 2,395–2,966 for a 32-char md5 column (8–10x); the
integer cost and the fixed overhead come out stable across delete densities
(288–299 and 537–598), which is what they should do, since decoding does not
depend on how much is deleted.
So the portable form of the threshold is **one delete check ≈ decoding
1.6–3.2 integer columns ≈ 0.16–0.39 of one md5 string column**, and "five
columns" is an artifact of this table's column order.
One caution on reading the sample counts as latency. Across all 24
configurations measured here, regressing the measured wall-clock reduction on
the sample-based scan reduction through the origin gives a slope of 0.53 (R² =
0.72): **only about half of a profiler-measured CPU saving arrives as wall
clock.** (Taking the ratio only where wall clock clears the noise band gives
1.73x, but that selects for large effects; the regression is the honest
summary.)
I have since established most of the mechanism, and it matters for how you
read every CPU number here. Splitting one run's process CPU by frame shows that
**JIT compilation is 41.6% of it** — these scans are short-lived JVMs, so a
large fixed cost sits outside the scan subtree and dilutes any saving inside
it. Amortising it by raising the iteration count tenfold moves the dilution
factor to 0.447, close to the measured 0.53. **That is a property of a
short-lived benchmark JVM, not of the workload**, which means the wall-clock
figures in this issue are conservative with respect to a long-lived executor.
Consistent with that, the one wall-clock result measured on a real cluster —
where executors live for the whole job — is a clean **−9.5%** with all six
paired rounds agreeing.
Combining the two — the change removes ~87% of the delete check, and about
half of that reaches latency — predicts that the effect clears my 9.7% noise
band once the delete check is above **~20% of scan CPU**. The ten-integer
projection sits at 21.5% share and measured −9.1%, right on the boundary. Every
scan-subtree percentage quoted in this issue is subject to this correction.
I also swept task parallelism to check how much that 0.53 depends on it,
since the threshold is derived from it. Fitting the correction factor against
`local[N]` for N = 1, 2, 4, 16 (96 profiles, one 16-vCPU instance, split size
pinned so task count scales) gives **1.30 × N^0.14** (R² = 0.95): 1.29x at N=1,
1.63x at N=4, 1.90x at N=16. So the factor does grow with parallelism — the
tail-task explanation is directionally supported — but only by 1.5x across a
16x range, which moves the threshold from about 18% to about 21%. The
break-even is **around a fifth of scan CPU** and does not change order of
magnitude with scale. Note that `local[N]` is a thread pool in one JVM; this
says nothing about shuffle or executor scheduling in a real cluster.
**Wall clock at 20 columns is not interpretable, in either direction.** The
three configurations came out +8.7% / −3.9% / −3.2% (patched vs current), all
inside the 9.7% run-to-run band and not even agreeing on sign. Pairing runs
within a round, the patched arm was slower in 13 of 18 pairs (sign test p ≈
0.10, not significant); before flipping the arm order it was 8 of 9, so part of
that was an order effect rather than the jar. I make no wall-clock claim at
this projection width.
So the honest end-to-end statement is: **the change reduces delete-check CPU
by 3.0x–9.3x across every projection I measured, and that turns into a visible
query-latency win while the delete check is above roughly a fifth of scan
CPU.** On this schema that means projections of up to about ten fixed-width
columns, or fewer than one md5 string column. Below that threshold the CPU
saving is real but I cannot measure it in query time.
</details>
<details>
<summary><b>Evidence — does this hold outside my machine?</b> (60
lines)</summary>
Everything above was measured on one developer machine. Since the
delete-check share is the number
the rest of the argument rests on, I re-ran the same two jars
(byte-identical, verified by md5) on
the same table bytes in three more environments. Delete-check share of scan
CPU, narrow projection:
| delete density | Zen 3 / WSL2 | Sapphire Rapids | Graviton3 | S3 instead
of local disk | page cache dropped |
|---|---|---|---|---|---|
| 0.5% | 43.2% | 49.3% | 48.4% | — | — |
| **6.1%** | **49.4%** | **56.2%** | **54.3%** | **58.0%** (vs 56.2% on EBS)
| **52.0%** (vs 51.0% warm) |
| 7.0% | 33.4% | 39.3% | 35.6% | — | — |
The share moves by at most a few points, and where it moves it moves *up*
off my machine, so the
numbers quoted above are the conservative end.
Two of those results were surprises worth stating, because both were
predictions I wrote down
first and got wrong:
- **Dropping the page cache does not lower the share, and neither does
reading from S3.**
`ctimer` samples CPU time, so I/O *wait* never enters the denominator; and
the S3 client's own
CPU (HTTP, TLS, checksums) turns out to be only ~1.4 percentage points of
total CPU here, far
too small to move it. Storage changes what fraction of wall clock is CPU,
not what fraction of
CPU is the delete check.
- **The patch helps *more* on newer cores**, not less: 11.3x–17.9x on the
cloud instances against
8.6x–9.3x on mine, because the per-row probe gets relatively more
expensive while the bulk range
traversal gets cheaper. I first attributed that to branch prediction.
Hardware counters do not
support it (see below), so I state it as an observation, not a mechanism.
**One thing does move the share, and it is not storage — it is file count.**
I split the same
32M rows into 4 files and into 488 files, choosing row counts so that each
file is a whole number
of 65,536-position chunks. That makes the two tables produce
**byte-identical Roaring containers**
(488 containers, `array:460 + bitmap:28`, 3,901,922 bytes, same deleted
rows) — the only
difference is whether those containers sit in 4 puffin files or 488.
| | delete-check share, EBS | delete-check share, S3 | wall clock |
|---|---|---|---|
| 4 files | 60.7% | 61.1% | 0.91 s |
| 488 files | **49.7%** | **49.0%** | 1.31 s |
The share falls by about 11 points, by the same amount on both storages.
Splitting the profile by
frame shows why: the per-row batch loop does not move (1,556 → 1,356 samples
— same rows, same
containers), while **DV load/deserialization grows 2.5x and non-DV scan work
grows 45%**. Those
are per-file fixed costs — footer parsing, one DV read per file — and they
enlarge the denominator.
Two consequences for reading this issue:
- The 43–58% band assumes a table with few, large files. On a table with
many small files, expect
the low end or below.
- **The speedup is diluted the same way**: 12.4x at 4 files against 6.1x at
488, on the same
dedicated instance. The patch is not worse — the loop it fixes is
unchanged — but the fixed cost
it does not touch takes a larger share of the denominator. I measured up
to 488 files; I have not
measured thousands.
One methodological note that cuts against my own earlier numbers: the
run-to-run spread on the
dedicated instances was **7–13%**, against **27% median** on my WSL2 box.
The 9.7% noise floor I
use throughout this issue is a property of my development environment more
than of the workload.
Anything I mark "inside the noise" might be resolvable on quieter hardware —
and in one case that
mattered: on my machine a shuffle-heavy query looked like the patch made it
*slower* (1 of 6 paired
rounds favouring the patch); on a dedicated instance with 12 rounds it is
**faster in 12 of 12**.
I was looking at a regression that was not there.
</details>
<details>
<summary><b>Evidence — what a user can do before this is fixed</b> (61
lines)</summary>
The delete check is cheap or expensive depending on which Roaring container
the positions land in,
so the physical layout of the table matters independently of this patch. I
tested that directly:
two tables with the **same rows deleted** (identical predicate, verified to
the row: 430,575 in
both), differing only in whether the table was sorted by the column the
deletes target. The scan
carries no predicate, so no files can be skipped — this isolates the
delete-check effect from the
file-pruning effect that sorting also gives you.
| | delete-check CPU | share of scan | vs baseline |
|---|---|---|---|
| unsorted, unpatched | 915 | 51.0% | 1.0x |
| **sorted, unpatched** | 327 | 27.3% | **2.8x** |
| unsorted, patched | 117 | 10.9% | 7.8x |
| **sorted + patched** | 74 | 8.0% | **12.4x** |
The mechanism is visible in the files: unsorted, every chunk is an `array`
container at ~7,000 bytes;
sorted, every chunk is a `run` container at 6-10 bytes, and the whole
deletion vector shrinks from
864 KB to 2.5 KB. Decoding cost is unchanged between the two (non-delete
scan samples 876 vs 883),
which is what makes the comparison valid.
**This has a condition, and it matters.** Sorting only helps when the
deletes concentrate on
relatively few distinct values of the sort key. Sweeping the key's
cardinality:
| sort key cardinality | rows per value | delete-check gain |
|---|---|---|
| ~1,000 | ~8,000 | **2.78x** |
| ~1,000,000 | ~8 | 1.10x |
| ~10^9 (effectively unique) | 1 | **0.89x** |
So "sort by the delete key" is good advice for deletes driven by date,
region or tenant, and no
advice at all for deletes driven by individual row ids — there the sorted
table performs like the
unsorted one and you have paid the sort for nothing. The two remedies also
overlap: the sorting
gain drops from 2.78x to 1.58x once this patch is applied, since the patch
has already removed most
of what sorting was saving.
(Wall-clock differences on this axis did not clear my noise band, so the
numbers above are CPU
samples, where the repeat ranges are disjoint.)
**What the sort costs.** I measured that too, because advice that only
prices the upside is not
advice. Writing the same 8M-row table sorted rather than unsorted, four
alternating rounds each:
| | write wall clock | stored bytes |
|---|---|---|
| unsorted | 20.07s (16.31-20.84) | 857,294,068 |
| **sorted** | **32.74s** (29.74-36.92) | **879,967,939** |
Writing costs **63% more** (+12.67s, or +1.58 µs per row; the ranges are
disjoint). Storage grows
**2.6%**, which surprised me — I had predicted it would shrink. Per-column
metrics explain it
exactly: the sort key itself collapses from 10,051,608 to **25,151 bytes**
(399x smaller, it becomes
runs), but `id` doubles (8.2 MB to 16.9 MB) and an id-derived column doubles
with it, because in the
unsorted table those were perfectly sequential within each file and
delta-encoded almost for free.
The other 17 columns move by less than 0.1%. **Sorting by a delete key
un-sorts whatever was
naturally ordered** — usually an auto-increment id or a load timestamp.
Putting cost and benefit together: at 0.042s saved per scan against 12.67s
of extra write, the sort
pays for itself after roughly **300 scans**. That ratio is fairly portable —
both sides scale with
row count, and the delete-check saving is independent of projection width —
so the practical rule is
*sort if the table is written once and read hundreds of times*, and don't if
it is rewritten often
by streaming upserts.
</details>
<details>
<summary><b>Evidence — a distributed cluster</b> (39 lines)</summary>
Everything above runs in one JVM, so I put the same two jars on a 3-node
Spark 4.0.4 standalone
cluster (1 master + 2 workers, 8 cores total, table on S3, profiler attached
to the *executors* via
`spark.executor.extraJavaOptions` and the per-executor profiles summed per
run).
| query | delete-check share of scan CPU | patch speedup | wall clock |
|---|---|---|---|
| scan, one JVM | 53.95% | 6.8x | — |
| **scan, cluster** | **41.58%** | **5.30x** | **−9.5%** (0/6 pairs, sign
test p = 0.03) |
| aggregation, one JVM | 43.19% ⚠️ | 6.9x | — |
| **aggregation, cluster** | **41.57%** | **5.07x** | +0.7% (inside noise) |
Two things worth stating, both of which contradict what I predicted:
- **The share drops to about 42% and the speedup to about 5x.** Executor
startup, task
serialization and S3 reads all land inside the scan subtree, enlarging the
denominator. The
delete check is still the single largest identified item, but the 58% top
of my range is a
single-JVM number and I have corrected the summary accordingly.
- **In the cluster the shuffle does not move the share** — 41.58% with no
shuffle against 41.57%
with a large one.
I originally contrasted this with a single-JVM pair of 53.95% against
43.19% and offered an
explanation for why the cluster behaved differently. **That contrast did
not survive
re-measurement.** The 43.19% came from my development machine, whose
round-to-round spread is
about three times that of a dedicated instance. Re-running the same three
queries on a dedicated
`m7i.xlarge` with 12 rounds instead of 6 gives **59.6% / 58.0% / 55.6%** —
a spread of 4.0
points, not 10.8. So the shuffle does not move the share in a single JVM
either, and there is
nothing left for the cluster to explain. I have removed the claim rather
than the data.
The wall-clock result is the useful one: on the plain scan the patched build
is **9.5% faster end
to end on the cluster**, with all six paired rounds agreeing in sign. That
is the first time I have
been able to claim a latency difference outside a single JVM.
Caveat on size: 8M rows over 8 cores is small enough that the cluster is
2.3x slower than one JVM
on the same query (0.50s vs 0.215s), which means fixed overhead is inflating
the denominator. On a
job large enough to be worth a cluster I would expect the share to move back
toward the single-JVM
figure, so 42% reads as a floor.
</details>
<details>
<summary><b>Evidence — hardware counters, and a mechanism I had wrong</b>
(65 lines)</summary>
I had assumed the per-row probe was expensive because the `array` container
does a binary search
whose branches mispredict. **That is not what the counters say.** Running
the microbenchmark under
`-prof perfnorm` (AMD Zen 3, 5000-row batch, 2 forks x 3 iterations):
| pattern | container | us/op | cycles/row | **insn/row** | branches/row |
**br-misses/row** | IPC |
|---|---|---|---|---|---|---|---|
| 0.5% density | array (card. 327) | 66.05 | 58.4 | 282.9 | 77.2 | 0.0126 |
4.84 |
| 5% density | array (card. 3,276) | 85.21 | 78.8 | 370.0 | 96.6 | 0.0763 |
4.70 |
| 12% density | **bitmap** | **21.13** | **18.6** | **86.5** | 17.8 | 0.0292
| 4.66 |
| scattered runs | run | 45.27 | 40.4 | 191.4 | 45.0 | 0.0066 | 4.74 |
Three things fall out:
- **Branch misprediction explains about 1.4% of the gap, not the gap.** To
account for the
301,045-cycle difference between the 5% and 12% patterns at an 18-cycle
penalty you would need
roughly 16,700 extra mispredicts per batch. There are 235. For the 0.5%
pattern the array
container actually mispredicts *less* than the bitmap one does.
- **Cache locality does not explain it either.** The fastest pattern
(bitmap) takes **8x more**
L1-dcache misses than the slow ones — 1,173.9 per op against 141.2.
- **IPC is flat at 4.66–4.84 across all four patterns**, and cycles track
instructions to within
0.8–3.8%. Nothing is stalling. The containers differ in **how many
instructions they execute
per row**, and that is the whole story.
This cuts for the proposal rather than against it. Even the cheapest
container costs **86.5
instructions per row**, because `RoaringPositionBitmap.contains(long)` has
to split the position,
binary-search the key array and follow two levels of indirection before it
can test a bit. That
fixed per-call cost is paid 5,000 times per batch whatever the container is.
The batch-oriented API
pays it a handful of times instead, which is why the speedups are as large
as they are.
Control: before trusting numbers this low I checked that the counters work
under this hypervisor,
with a sorted-vs-shuffled branch experiment — 9.2M vs 108.6M mispredicts
(11.8x), matching the
theoretical 50% miss rate to within 3%.
**The same holds in a real Spark scan, but with two qualifications.**
Wrapping `spark-submit`
itself in `perf stat` (3 alternating rounds):
| arm | cycles | instructions | branch-misses | miss rate | IPC |
|---|---|---|---|---|---|
| baseline | 237,714,166,096 | 390,468,338,748 | 1,204,681,888 | 1.55% |
1.64 |
| patched | 203,636,206,746 | 260,507,995,090 | 1,200,937,204 | 2.51% | 1.28
|
- **Branch misprediction explains 0.2% of the patch's effect**, and the
clearest sign is that the
absolute count barely moves (1.2047G to 1.2009G) even though the patch
deletes **38% of all
branches**. What it removes are branches that were already being predicted
correctly.
- **But instruction count does not fully explain it either, unlike in the
microbenchmark.** The
patch removes 50% of instructions and 17% of cycles; IPC falls from 1.64
to 1.28. The work that
remains is more stall-bound. So the honest statement for a real query is
*"removing half the
instructions buys you a sixth of the time"*, which is also why my
CPU-sample numbers overstate
wall-clock gains by about 1.9x.
- The real scan's branch miss rate is **1.55%**, roughly 100x the
microbenchmark's. Microbenchmark
counter values do not transfer; only the conclusion does.
One useful side effect: profiling with `cycles` instead of the CPU-time
timer gives a delete-check
share of **52.2%** against **50.0%** from CPU-time sampling — within 2.2
points. Every share figure
in this issue holds under cycle-accurate attribution too.
**Where branch prediction does matter.** At very high delete densities the
row test
`if (!contains(pos))` becomes a coin flip. Comparing 8% against 50% density,
mispredicts attributed
to the delete-check region rise from 82.7M to 115.5M (**+40%**, disjoint
ranges), and the region's
share of scan mispredicts goes 41.1% to 51.6%. That accounts for about
**28%** of the extra cycles
the delete check spends there — small in absolute terms, but 20x more than
in any other comparison
I ran. I have not identified the remaining 72%.
</details>
<details>
<summary><b>Evidence — microbenchmark</b> (28 lines)</summary>
JMH, batchSize 5000, positions built into a full 65536-position chunk at the
target density, then round-tripped through
`BitmapPositionDeleteIndex.serialize()`/`deserialize()` so the containers match
what the read path actually sees. Time per batch, µs (RoaringBitmap 1.6.20
standalone):
| implementation | sparse 0.5% | medium 5% | dense 12% | run (L=64, 5%) |
empty batch |
|---|---|---|---|---|---|
| current (`isDeleted` per row) | 70.37 | 81.54 | 19.17 | 44.67 | 23.31 |
| `forEachInRange` + gap fill | **0.47** | **1.86** | **3.38** | **0.80** |
**0.26** |
| `forAllInRange` + `RelativeRangeConsumer` | 0.47 | 1.81 | 2.95 | 0.30 |
0.27 |
The `run` column uses length-64 deleted blocks spread across the chunk to
reach 5% density
(51 runs), so the batch partially overlaps the deletes. A fully-deleted
batch would report
~3800x, which is a degenerate best case and not a number worth quoting.
Two things fall out of the `run` column:
- At **identical density (5%)**, the array container costs 81.54 µs and the
run container
44.67 µs — 1.83x, purely from container structure. `log2(3277)/log2(51) =
2.06`, so the
binary-search model predicts this within 11%. Table layout changes what
the reader pays
before any code change does.
- **`forAllInRange` is 2.7x faster than `forEachInRange` on run containers**
(0.30 vs 0.80),
because it can skip the gaps between runs with `acceptAllAbsent` instead
of only being told
where the deletes are. On array and bitmap containers the two are
equivalent. Both are still
≥55x over the current code, so I lean towards the smaller API, but this is
the concrete cost
of that choice and I am happy to go the other way.
All variants were asserted to produce byte-identical `rowIdMapping` and live
counts before measurement. The end-to-end gain is smaller than the
microbenchmark ratio because the delete-check subtree also contains DV
deserialization, the mapping array write, and the `RoaringPositionBitmap`
wrapper — none of which this change touches.
</details>
<details>
<summary><b>Evidence — the other two paths</b> (33 lines)</summary>
**`buildIsDeleted`** is reached when a query projects the `_deleted`
metadata column. Same tables,
same method, only the projection differs:
| configuration | current | patched | reduction |
|---|---|---|---|
| 0.5% deletes + `_deleted` | 649 (573–717) | **34 (25–42)** | **18.9x** |
| 6.1% deletes + `_deleted` | 854 (793–892) | **57 (56–58)** | **14.9x** |
The gain is larger here than for `buildRowIdMapping` because this path only
has to mark the
deleted positions — it never fills the gaps with live row ids, so the range
traversal does
~25 writes per batch instead of ~4,975. Row counts and the number of rows
flagged as deleted
were identical between the two jars (40,154 and 488,161, matching the DV
cardinalities exactly).
**Equality deletes.** I built two tables with equality delete files written
through
`Parquet.writeDeletes(...).buildEqualityWriter()` and committed with
`RowDelta`, since Spark
itself only writes position deletes:
| configuration | current | patched | ratio | verdict |
|---|---|---|---|---|
| equality deletes only | 3,303 (2,590–4,249) | 2,635 (2,594–2,679) | 1.25x
| ranges overlap — no claim |
| DVs **+** equality deletes | 3,362 (3,177–3,559) | 3,524 (3,267–3,955) |
0.95x | ranges overlap — no claim |
No regression. The second table is the case where the gate actively refuses
the fast path, and
the only added work there is one `hasEqDeletes()` call per batch.
The first table turns out to be a useful negative control: with no position
deletes,
`deletedPositions == null` short-circuits before `hasEqDeletes()` is even
called, so **both jars
execute byte-identical code** — and the measurement still differed by 1.25x.
That is a direct
read on how much apparent effect noise can produce at these sample counts,
and it is an order of
magnitude below the numbers claimed above.
</details>
### Proposal
Add a range-scoped traversal to `PositionDeleteIndex` with a default
implementation that preserves current behavior, so no existing implementation
has to change:
```java
/**
* Traverses the deleted positions within the given range in ascending
order, applying the
* provided consumer.
*
* @param posStart the first position in the range, inclusive
* @param length the number of positions in the range
* @param consumer a consumer for the deleted positions in the range
*/
default void forEachInRange(long posStart, int length, LongConsumer
consumer) {
for (int index = 0; index < length; index++) {
long pos = posStart + index;
if (isDeleted(pos)) {
consumer.accept(pos);
}
}
}
```
Override it in `BitmapPositionDeleteIndex`, delegating through
`RoaringPositionBitmap` to `RoaringBitmap.forEachInRange`.
`EmptyPositionDeleteIndex` overrides it to a no-op.
`RoaringPositionBitmap.forEachInRange` has to split the range across the
32-bit keys, but since `length` is an `int` and a single key covers 2³²
positions, a range can span at most two keys.
Then in `ColumnarBatchUtil`, take the range path when there are no equality
deletes and fall back to the existing per-row loop otherwise:
```java
PositionDeleteIndex deletedPositions = deletes.deletedRowPositions();
if (deletedPositions != null && !deletes.hasEqDeletes()) {
return buildRowIdMapping(deletedPositions, deletes, rowStartPosInBatch,
batchSize);
}
// existing per-row loop
```
**Note the condition is `!deletes.hasEqDeletes()`, not `eqDeleteFilter ==
null`.** `DeleteFilter.eqDeletedRowFilter()` returns `t -> true` rather than
`null` when there are no equality deletes (`DeleteFilter.java:245`), so a null
check would never take the fast path.
**The equality-delete case must keep the per-row loop**, because
`eqDeleteFilter.test(row)` needs each row and `ColumnarBatchRow.rowId` is
advanced inside that loop. The range path only applies to position deletes /
deletion vectors — the V3 DV case.
`deletes.incrementDeleteCount()` is still called once per deleted row.
I kept the proposed API deliberately small. `RelativeRangeConsumer` (which
additionally reports absent positions and has bulk
`acceptAllPresent`/`acceptAllAbsent` callbacks) is faster in the dense and
fully-deleted cases, but it is a much larger API surface to expose through
`PositionDeleteIndex`. It also turns out that `RoaringBitmap.forEachInRange`
internally delegates to `forAllInRange` with an
`IntConsumerRelativeRangeAdapter`, so the small API already gets most of the
benefit. Happy to go the other way if maintainers prefer it.
### Tests
`TestColumnarBatchUtil` currently mocks `PositionDeleteIndex` and stubs only
`isDeleted`. A Mockito mock does nothing for default methods, so the
position-only tests would see "nothing deleted" once the range path is taken. I
changed those tests to use a real index built with
`Deletes.toPositionIndex(...)`, which also makes them exercise the actual
bitmap. Tests that combine position and equality deletes additionally need
`hasEqDeletes()` stubbed — they currently stub `hasPosDeletes()`, which
`ColumnarBatchUtil` does not read.
New coverage on the branch:
- `TestPositionDeleteIndexForEachInRange` (core, 12 tests): empty index,
zero/negative length, inclusive-exclusive boundaries, ascending order, ranges
spanning a 65536-position Roaring chunk, ranges spanning a 32-bit key boundary,
ranges past the allocated bitmaps, run containers after `runLengthEncode`, and
randomized comparison of 500 ranges against a per-position `isDeleted` scan —
for both the bitmap implementation and the interface default.
- `TestColumnarBatchUtil` (spark, 5 new tests): non-zero batch start
positions, deletes entirely outside the batch range, all rows deleted by
position, and a randomized comparison of 60 batches against the per-row path
including the delete-counter call count.
```
:iceberg-core:test --tests "org.apache.iceberg.deletes.*" 55 tests, 0
failures
:iceberg-spark:iceberg-spark-4.0_2.13:test --tests "...TestColumnarBatchUtil"
17 tests, 0
failures
```
### Compatibility
- `PositionDeleteIndex` already extends via `default` methods (`merge`,
`forEach`, `cardinality`, `serialize`), so this follows an established pattern
and is source- and binary-compatible for external implementations.
- The default implementation is exactly the current behavior, so correctness
does not depend on any implementation adopting the override.
- No format or spec change.
### Caveats on the numbers
- The primary measurements are on a developer machine (WSL2), Spark
`local[4]` mode; the portability checks above are on dedicated EC2 instances.
**Correction to an earlier version of this text:** I had written that no
hardware PMU was available and that cycle- and branch-level attribution would
need a bare-metal run. That was wrong — recent WSL2 kernels expose a Hyper-V
vPMU, and the counter evidence in *hardware counters* above was collected with
it. I had asserted the limitation in eight places over eight months without
checking it.
- Most of the numbers are single-JVM `local[N]`. Adding a `GROUP BY` moves
the delete check from **14.0% of total query CPU to 6.8% and then 2.4%** as the
aggregation grows, while its share *of the scan* stays in the 43-54% band and
the absolute CPU saved is unchanged. I also ran it on a 3-node Spark standalone
cluster (see below); the cluster there is small (8 cores, 8M rows), so fixed
overhead dominates and the cluster is actually 2.3x *slower* than one JVM on
the same query. A production-sized job would sit closer to the single-JVM
numbers, so treat 42% as a lower bound rather than a typical cluster value.
- One table shape for most axes: 8M rows, 4 files, 20 columns. File-skipping
predicates and nested types are unmeasured. Many-file tables **are** now
measured (4 vs 488 files, see above) but only to 488; thousands of files, and
cross-region S3, remain unmeasured.
- The run-to-run spread of the same configuration is **9.7% median, 25.9%
max** across 15 configurations × 3 runs. I report ranges and do not claim
differences inside that band. The patched arm has larger *relative* spread (up
to 59%) simply because its absolute sample counts are small (43–117).
- Equality deletes were written on a single column (`id`). Multi-column
equality deletes were not measured.
- I ran the length-controlled axis afterwards (four md5-derived string
columns of 8, 16, 24 and 32 characters, same type, one column projected at a
time). Fitting scan samples against length gives **1,099 + 63.8 x length** (R²
= 0.94), so it is neither purely "because it is a string" nor purely "because
it is wide" — both terms are real. At 8 characters the fixed per-column term
and the length term are about equal; at 32 characters length dominates 2:1.
Against an integer column the same measurement gives 2.1x for an 8-char string
and 4.2x for a 32-char one, so the "8-10 integer columns" figure above applies
to full-length md5, not to short strings. One point (24 chars) appeared to sit
off the line, but that was an artifact of my own aggregation: I had averaged
six rounds, and one round was disturbed (three of the four string columns
spiked 1.3-1.8x in it, and its wall clock spiked with them). Taking medians
instead, as the rest of my tooling does, gives **1,808 + 61.6 x length
** with R² = 0.95 and leaves the 24-char point 6% off the fit, inside my 9.7%
noise band. There is no anomaly to explain.
- The 0.53 sample-to-wall-clock slope is a fit to data I had already
collected, not a prediction I tested. It is measured on one machine at
`local[4]` with a warm page cache; I would expect a different slope at other
parallelism or on other storage. Its main driver is JIT compilation in a
short-lived JVM (41.6% of process CPU), which means it understates what a
long-lived executor would see.
- The 1-column wall-clock rows come from 3 repetitions; every other row from
6.
- The `_deleted` projection emits every row rather than filtering, so its
wall clock is not comparable to the plain scans above.
- The delete-check share depends on file count as well as projection width:
60.7% at 4 files against 49.7% at 488, with containers held byte-identical.
Quote the band with a file-layout assumption attached.
- Two numbers in this issue were withdrawn after re-measuring on quieter
hardware: a single-JVM "aggregation lowers the share to 43.19%" contrast (it
does not — 4.0 points across three queries on a dedicated instance) and an
apparent wall-clock regression on shuffle-heavy queries (the patch is faster,
12 of 12 paired rounds). Both originated on a machine whose spread is ~3x that
of a dedicated instance.
- The microbenchmark reuses the output buffer to isolate the algorithm. The
real code allocates `new int[batchSize]` per batch.
### Prior work
I looked for existing discussion before filing: #12053 and #12054 touch
`ColumnarBatchUtil` but are Javadoc and unit tests. I did not find an existing
issue on the per-row probing cost, nor any reference to the Roaring range APIs
in this repository.
### Query engine
Spark
### Willingness to contribute
- [x] I can contribute this improvement/feature independently
- [ ] I would be willing to contribute this improvement/feature with
guidance from the Iceberg community
- [ ] I cannot contribute this improvement/feature at this 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]
