[Spark] Capture and emit aggregated data-file (Parquet footer) metrics at commit time
- Dominant language
- Java
- Stars
- 9.2k
- Forks
- 3.5k
- Avg merge
- 2d 11h
- Merged PRs (30d)
- 132
Description
### Proposed Change
# Proposal: Capture and emit aggregated data-file (Parquet footer) metrics at commit time
### Feature Request / Improvement
Provide an opt-in mechanism to capture aggregated physical/storage statistics directly from Parquet footers during a write, and surface them through Iceberg's existing event framework at commit time — **without** persisting them in table metadata.
### Motivation
Iceberg infers column-level metrics (`value_counts`, `null_value_counts`, `nan_value_counts`, `lower_bounds`, `upper_bounds`) and stores them per-`DataFile` in manifests. Because these metrics live in manifest entries, collecting them for every column of a wide table bloats manifests and slows scan planning. To bound this, Iceberg caps inferred metrics at the first N columns:
```write.metadata.metrics.max-inferred-column-defaults = 100 (default)```
This cap is the right trade-off for *query-planning* metadata, but it creates a gap for **observability / storage-health** use cases:
1. **Wide tables (>100 columns)** get no inferred metrics for columns beyond the cap unless each is configured explicitly. There is no aggregate visibility into the storage footprint of those columns.
2. **Physical/storage statistics aren't captured at all** in metadata — e.g. per-column compressed vs. uncompressed size, encoding/codec usage, dictionary/bloom-filter presence, and file-size distribution (percentiles, small-file counts). Today the only way to get these is to re-open and re-read files after the fact, which is expensive and runs on stale data.
3. Raising the metrics cap to get more coverage directly **worsens metadata bloat and planning latency** — the exact problem the cap exists to prevent.
The key insight: metrics needed for *query planning* and metrics needed for *operational observability* have different lifetimes and storage requirements. Observability metrics don't need to live in manifests forever; they're most valuable **at write time**, aggregated, and emitted to a monitoring system. Decoupling the two lets us collect rich, uncapped, aggregate statistics cheaply without touching the metadata hot path.
### Proposed Solution
Add an opt-in write-path feature that, during a Spark write:
1. **Reads Parquet footers** for the data files produced by each task (footers are already on the writer; this avoids a separate read pass).
2. **Aggregates statistics map-side**, grouped by partition and by column field-id, so per-file payloads never accumulate on the driver.
3. **Bridges the aggregated result into the commit** and publishes it as a new event through `org.apache.iceberg.events.Listeners`, so any deployment can register a listener and route the metrics to its own monitoring/observability system.
Because the metrics flow through the event framework rather than into `TableMetadata`/manifests, there is **no metadata bloat and no column cap**, and the commit path's persisted state is unchanged.
### High-level implementation details
**Executor-side capture (Spark write path)**
- A new opt-in table property gates the behavior; when disabled (the default), there is zero overhead.
- After each task's `DataWriter.commit()`, the produced `DataFile`s' Parquet footers are read via `FileIO`/`InputFile` (with a small adapter from Iceberg's `SeekableInputStream` to Parquet's `InputFile`).
- For each footer, statistics are extracted per column chunk and mapped back to **Iceberg field-ids** (via the Parquet schema's field ids, so the data is schema-evolution safe), then folded into a partition-keyed aggregate.
**Map-side aggregation via a Spark `AccumulatorV2`**
- Aggregates are accumulated into a custom accumulator and merged on the driver. To avoid driver heap pressure on writes producing very large numbers of partitions, the accumulator:
- merges executor-local aggregates directly (no per-file objects retained), and
- returns an empty snapshot from `value()` so Spark does not retain `O(numTasks)` deep copies in `TaskInfo` accumulables; the real data is consumed once, at commit, via an explicit drain.
**Commit-time emission**
- Just before `commitOperation()`, the merged aggregate is finalized (file-size percentiles, small-file counts) and handed to the commit via a thread-local bridge, then published as an event during the synchronous commit notification.
- The bridge is cleared on commit success, on `CommitStateUnknownException`, and on any other failure, so nothing leaks across commits.
- The core event type stays free of Parquet/Spark types (the payload is carried in a serialized, engine-agnostic form), keeping the engine boundary clean.
**Metrics captured (aggregated per partition + per column)**
- Row counts and total record counts.
- File counts, total file size, and a file-size distribution: average, p50/p75/p90/p95/p99, fixed size buckets, and a configurable small-file count.
- Per-column (by field-id): compressed size, uncompressed size, value count, null count, codec (or "mixed"), and dictionary-page / bloom-filter presence.
**Configuration (new table properties)**
```write.data-file-metrics.enabled (default: false) ```
```write.data-file-metrics.small-file-threshold-bytes (default: 128 KB)```
### Why the event framework instead of metadata
- No manifest/`TableMetadata` changes → no planning-path impact and no spec change.
- No column cap → aggregate coverage for arbitrarily wide tables.
- Metrics are available in **real time** at commit, computed from footers already produced by the write.
- Fully opt-in and pluggable: deployments that don't register a listener pay nothing.
### Scope / non-goals
- Initial implementation targets the **Spark** write path (`SparkWrite`); Flink and others could follow the same pattern.
- Parquet only initially (footer-based); ORC/Avro could be added later.
- This does **not** replace Iceberg's existing inferred metrics used for pruning — it's a complementary, observability-focused channel.
### Open questions for discussion
1. Is the **events framework** the right surface, or should this be a dedicated metrics/reporting interface (similar in spirit to `MetricsReporter` for scans, but for writes)?
2. Naming/placement of the new property and event type.
3. Should the aggregation granularity (per-partition, per-column) and the file-size bucket boundaries be configurable?
4. Appetite for extending beyond Parquet, and beyond Spark, in follow-ups.
### Proposal document
_No response_
### Specifications
- [x] Table
- [ ] View
- [ ] REST
- [ ] Puffin
- [ ] Encryption
- [ ] Other
Contributor guide
Assessment
This issue has not been assessed yet.