Sort/z-order compaction permanently no-ops on size-healthy but badly-clustered tables — no way to detect or select by overlap
- Dominant language
- Java
- Stars
- 9.2k
- Forks
- 3.5k
- Avg merge
- 2d 11h
- Merged PRs (30d)
- 132
Description
### Feature Request / Improvement
**TL;DR: `rewrite_data_files` candidate selection only looks at file size, so a table whose files are all within the healthy size band is *permanently* excluded from sort/z-order compaction. The job reports success on every run while sort-key overlap between files never improves — and there is currently no way to even detect this state.**
### Mechanism
File selection for all rewrite strategies lives in `SizeBasedFileRewritePlanner#outsideDesiredFileSizeRange`:
```java
protected boolean outsideDesiredFileSizeRange(T task) {
return task.length() < minFileSize || task.length() > maxFileSize;
}
```
`SparkShufflingDataRewritePlanner` (sort/z-order) extends `BinPackRewriteFilePlanner` and inherits this selection unchanged — sort key overlap between files is never considered. So:
- files inside `[min-file-size-bytes, max-file-size-bytes]` → never candidates, no matter how badly they overlap on the sort key
- the rewrite completes with `rewritten_data_files_count = 0` and reports **success**
The perverse consequence: a *fragmented* table (undersized files) gets selected and accidentally comes out perfectly clustered, while a *size-healthy* table accumulates overlap indefinitely. On one of our production tables, scheduled daily sort compaction "succeeded" every day while max overlap depth (number of file ranges covering a single point of the sort key) kept growing by ~1 per day.
### Deterministic reproduction (Iceberg 1.10.0, Spark 3.5, local)
Two tables with identical, fully-overlapping data — only file size differs:
| case | file sizes | rewrite result | overlap depth |
|---|---|---|---|
| A: size-healthy | 8 × ~2.3 MB (inside band) | `rewritten=0`, success | 8 → **8** (unchanged) |
| B: fragmented (control) | 8 × ~0.2 MB (below min) | `rewritten=8` | 8 → **1** |
repro.py (self-contained, ~2 min on a laptop)
```python
"""
Reproduction: sort compaction permanently no-ops on size-healthy tables.
Candidate selection in RewriteDataFiles considers only file size
(min-file-size-bytes / max-file-size-bytes / delete counts). A table whose
files are all within the healthy size band is never selected for rewrite,
even when every file overlaps every other file on the sort key. The sort
strategy therefore reports success daily while clustering never improves.
Control group: the same data written as undersized files IS selected and
gets perfectly clustered - fragmentation accidentally heals, health
permanently rots.
Run: .venv/bin/python repro.py
"""
import shutil
import tempfile
from pyspark.sql import SparkSession
ICEBERG = "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.10.0"
WAREHOUSE = tempfile.mkdtemp(prefix="iceberg-overlap-repro-")
TARGET_MB = 2 # scaled-down target-file-size; selection logic only uses ratios
NUM_FILES = 8
ROWS_PER_FILE_HEALTHY = 90_000 # ~2 MB parquet each -> inside [0.75t, 1.8t]
ROWS_PER_FILE_SMALL = 8_000 # ~0.3 MB each -> below min threshold
spark = (
SparkSession.builder.appName("overlap-repro")
.config("spark.jars.packages", ICEBERG)
.config("spark.sql.extensions",
"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
.config("spark.sql.catalog.local", "org.apache.iceberg.spark.SparkCatalog")
.config("spark.sql.catalog.local.type", "hadoop")
.config("spark.sql.catalog.local.warehouse", WAREHOUSE)
.config("spark.driver.memory", "2g")
.master("local[4]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
def create_table(name: str) -> None:
spark.sql(f"DROP TABLE IF EXISTS local.db.{name}")
spark.sql(
f"""CREATE TABLE local.db.{name} (id BIGINT, payload STRING)
USING iceberg
TBLPROPERTIES ('format-version'='2')"""
)
def write_overlapping_files(name: str, rows_per_file: int) -> None:
"""Each commit writes one file spanning the FULL id range -> total overlap.
The sort order is declared AFTER the appends (see main flow): declaring it
before would make appends locally sorted, which is not the scenario -- the
scenario is an existing unsorted table that sort compaction should fix.
"""
for i in range(NUM_FILES):
df = (
spark.range(rows_per_file)
.selectExpr(
# deterministic pseudo-random spread over the full key range
f"(id * 2654435761 + {i}) % 10000000 AS id",
"repeat(uuid(), 2) AS payload",
)
.coalesce(1)
)
df.writeTo(f"local.db.{name}").append()
def file_stats(name: str):
rows = spark.sql(
f"""SELECT file_size_in_bytes,
readable_metrics.id.lower_bound AS lo,
readable_metrics.id.upper_bound AS hi
FROM local.db.{name}.files"""
).collect()
return [(r.file_size_in_bytes, r.lo, r.hi) for r in rows]
def overlap_depth(intervals) -> int:
"""Max number of file [lo, hi] ranges covering a single point (sweep line)."""
events = []
for _, lo, hi in intervals:
events.append((lo, 1))
events.append((hi, -1)) # closed intervals: end after start at same key
events.sort(key=lambda e: (e[0], -e[1]))
depth = best = 0
for _, delta in events:
depth += delta
best = max(best, depth)
return best
def rewrite_sort(name: str):
return spark.sql(
f"""CALL local.system.rewrite_data_files(
table => 'db.{name}',
strategy => 'sort',
options => map('target-file-size-bytes', '{TARGET_MB * 1024 * 1024}')
)"""
).collect()[0]
def report(name: str, label: str) -> None:
stats = file_stats(name)
sizes = sorted(s // 1024 for s, _, _ in stats)
print(f" [{label}] files={len(stats)} sizes_kb={sizes} "
f"overlap_depth={overlap_depth(stats)}")
print(f"warehouse: {WAREHOUSE}\n")
# --- Case A: size-healthy files, fully overlapping ------------------------
print("Case A: size-healthy files (inside [min,max] band), fully overlapping")
create_table("healthy")
write_overlapping_files("healthy", ROWS_PER_FILE_HEALTHY)
spark.sql("ALTER TABLE local.db.healthy WRITE ORDERED BY id")
report("healthy", "before")
res = rewrite_sort("healthy")
print(f" rewrite result: rewritten={res.rewritten_data_files_count} "
f"added={res.added_data_files_count}")
report("healthy", "after ")
# --- Case B: control - same data as undersized files ----------------------
print("\nCase B (control): undersized files, fully overlapping")
create_table("fragmented")
write_overlapping_files("fragmented", ROWS_PER_FILE_SMALL)
spark.sql("ALTER TABLE local.db.fragmented WRITE ORDERED BY id")
report("fragmented", "before")
res = rewrite_sort("fragmented")
print(f" rewrite result: rewritten={res.rewritten_data_files_count} "
f"added={res.added_data_files_count}")
report("fragmented", "after ")
print("""
Expected outcome:
Case A: rewritten=0, depth unchanged (=8). Job reports success, sort order
never materializes, and this repeats on every scheduled run.
Case B: rewritten=8, depth -> 1. Being undersized accidentally heals it.
""")
spark.stop()
shutil.rmtree(WAREHOUSE, ignore_errors=True)
```
[repro.py](https://github.com/user-attachments/files/30642900/repro.py)
### Prior discussion
- In #12761, @pvary asked exactly the missing piece — *"Do you have a way to identify the non-sorted files somehow?"* — and the thread went stale without an answer. This proposal is intended as that answer.
- #2609's description already sketched overlap-aware selection ("find files A, B, C with significant overlap and rewrite them together"), but it never landed in selection logic.
- Recent work (#16827 Hilbert clustering, #16305 k-way merge) improves how data is *rewritten*, but not *which* files get selected — the selection gap applies to those strategies as well.
### Proposal
1. **Measurement first**: expose per-partition sort-key overlap depth computed purely from manifest `lower_bounds`/`upper_bounds` (no data file reads) — as a metadata table column or a procedure. Today every size-based health metric reports a badly-clustered table as perfectly healthy.
2. **Selection (opt-in)**: a planner option (e.g. `min-overlap-depth`) so sort/z-order strategies can pick up well-sized but badly-clustered files. The `filterFiles`/`filterFileGroups` extension points seem to accommodate this without changing default behavior.
Happy to work on a PR for (1) if maintainers think either direction is worth pursuing.
### Query engine
Spark
### Willingness to contribute
- [ ] I can contribute this improvement/feature independently
- [x] I would be willing to contribute this improvement/feature with guidance from the Iceberg community
- [ ] I cannot contribute this improvement/feature at this time
Contributor guide
Research direction
Start with SizeBasedFileRewritePlanner#outsideDesiredFileSizeRange and trace how SparkShufflingDataRewritePlanner selects files; run the provided repro.py against Iceberg 1.10.0 to confirm the no-op. Review the filterFiles/filterFileGroups extension points and existing rewrite-planner tests. Done should include a documented overlap measurement or opt-in selection path, tests for well-sized overlapping files, and unchanged default behavior.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, python, spark
- Domain
- data-engineering, databases
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100