rahil-c opened a new pull request, #19616: URL: https://github.com/apache/hudi/pull/19616
### Change Logs The Record Level Index lookup phase of a write is currently unobservable. `HoodieMetadataMetrics` has declared metric names since 0.7.0 (`lookup_record_index_time`, `lookup_record_index_key_count`, `lookup_record_index_key_hit_count`) that nothing references, and `HoodieBackedTableMetadata#readRecordIndexLocationsWithKeys` carries a standing `TODO [HUDI-9544]: Metric does not work for rdd based API due to lazy evaluation`. The counts exist as locals on the executor, but the lookup returns only hits — a miss produces no output row — so the driver cannot recover the denominator from the resulting RDD. This adds, behind an off-by-default config, per-commit counts for that phase: shards read, log files read, bytes resident in those shards, keys submitted and keys hit. They are published to the existing `MetricsReporter` and persisted to commit metadata. **Collection lives in `hudi-common`, at the method every engine already calls.** Both `SparkMetadataTableGlobalRecordLevelIndex` and Flink's `GlobalRecordLevelIndexBackend` call `readRecordIndexLocationsWithKeys`, so instrumenting there means one definition of every count and no duplicated implementation when Flink support follows. Transport is necessarily engine-specific — Spark aggregates through an `AccumulatorV2` drained on the driver; Flink would report per subtask through a `MetricGroup`. **The interface addition is a `default` method, not abstract.** `HoodieTableMetadata` has implementations outside this repository and an abstract method would break their compilation. As a `default` the change is purely additive: `FileSystemBackedTableMetadata` and `NoOpTableMetadata` are untouched, and only the record-index-backed implementation overrides it. **The accumulator is keyed by file group id and merged per key by field-wise `max`.** That makes collection idempotent under task retry, speculation and RDD recomputation — re-reporting a shard replaces its entry instead of adding to a running total. Keying on shard index would not work: the partitioned record index resolves that index against a single data table partition's slices, so two partitions both produce index 0 for two different file groups. `max` is also commutative and associative, which the accumulator merge contract requires. **Deliberately not included:** per-record insert/update counts, which commit metadata already carries as `numInserts`/`numUpdateWrites`; any histogram; and any metric label carrying a shard id, which would be a Prometheus cardinality blowup. Note that `bytes_in_shards_read` is the **footprint** of the shards touched, an upper bound rather than measured I/O — record index lookups push the key set into the file reader, so only a fraction of each shard is read. The name says so and the config documents it. ### Impact Off by default (`hoodie.metadata.record.index.lookup.stats.enable=false`). When disabled the collector is a no-op singleton, nothing is allocated on the executor path, and behaviour is unchanged. When enabled the cost is a handful of counters per shard actually read, plus one `extraMetadata` key per commit holding compact JSON. Gauge values are all `Long` and names go through `getMetricsName`, so they carry the configured reporter prefix — Datadog and Prometheus both cast gauge values to a number, and unprefixed names would collide across tables sharing a backend. ### Risk level: low Off by default and additive. The `hudi-common` change is one `default` interface method plus threading a collector through the record index read path. ### Documentation Update New config `hoodie.metadata.record.index.lookup.stats.enable` carries full documentation including the footprint-vs-IO caveat. ### Known gaps — this is a draft - **Counts can leak from an aborted commit into the next one.** The accumulator drains at `preCommit`; a commit that fails earlier leaves its counts in place. Needs draining on abort and on write-client close. - **No integration coverage for the partitioned-RLI path.** It is wired, but only the global path has an end-to-end test. - **`lookupMillis` is collected but not emitted.** Intended as a follow-up along with a skew summary (`max_keys_on_shard`, hottest shard), which the shard-keyed map already makes free to derive. - **Flink support is designed for but not implemented.** - Minor: `getLogFiles().count()` should be `getLogFileCount()`, and `getTotalFileSize()` can return `-1` for unknown-size files, which should be clamped. ### Contributor's checklist - [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute) - [x] Change Logs and Impact were stated clearly - [x] Adequate tests were added if applicable - [ ] CI passed 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
