rahil-c opened a new pull request, #19575: URL: https://github.com/apache/hudi/pull/19575
### Describe the issue this Pull Request addresses `HoodieMetadataMetrics` has declared three record-index lookup metrics for some time: ```java public static final String LOOKUP_RECORD_INDEX_TIME_STR = "lookup_record_index_time"; public static final String LOOKUP_RECORD_INDEX_KEYS_COUNT_STR = "lookup_record_index_key_count"; public static final String LOOKUP_RECORD_INDEX_KEYS_HITS_COUNT_STR = "lookup_record_index_key_hit_count"; ``` Nothing in the repository references any of them, and `HoodieBackedTableMetadata#readRecordIndexLocationsWithKeys` records why: ```java // TODO [HUDI-9544]: Metric does not work for rdd based API due to lazy evaluation. ``` So with RLI enabled there is currently no way to tell how many keys were looked up, how many hit the index, or how many shards were read — which is often the most expensive phase of an upsert. The numbers already exist. `keysToLookup.size()` and the size of the returned map are local variables inside `RecordIndexFileGroupLookupFunction.call()`. But the function returns **only the hits** — a miss produces no output row — so the driver cannot recover the denominator from the resulting RDD at any price. Recovering `records_looked_up` or `shards_read` would each cost an extra Spark job over data already computed, and caller attribution is not reconstructible after the fact at all. Executors could always count; they had no way to report. Relates to HUDI-9544 and #19063. ### Summary and Changelog Four counters are emitted per RLI shard and published at each commit, so an operator can see the index hit rate, how much of the index was read, and whether the index is earning its keep. **`fix(metrics)`: harden `DistributedRegistry`** — these registries live in process-wide static maps that outlive the SparkContext, the write client and the table. - `set()` is last-writer-wins, which is neither commutative nor associative; inside an `AccumulatorV2` the driver merges executor copies in an unspecified order, so an executor-side `set()` is non-deterministic. Now rejected from inside a task. No production caller exists today. - After a SparkContext restart in the same JVM (shells, notebooks, Spark Connect) the cached registry stayed bound to the dead context and executor updates silently stopped arriving. `isRegistered()` cannot detect this — it consults a Spark-global weak-reference table that `SparkContext.stop()` never clears, and the static map pins the registry so it is never collected. The application id is now stamped at registration and compared on lookup; on mismatch the entry is evicted and a **fresh** instance built, since `AccumulatorV2.register()` throws if the accumulator already carries registration metadata. Evict-and-recreate is atomic, or racing callers leave two live accumulators for one metric name with only one reachable. - `HoodieWrapperFileSystem` counters leaked into the next write client for the same table; now cleared on create and on close. **`feat(metrics)`: record and publish the counters** — `records_looked_up`, `hits`, `misses`, `shards_read`, each tagged by caller so tag-location traffic is distinguishable from the read client's dedupe traffic. The registry is resolved on the driver and captured by the lookup closure, so it rides inside the closure and nothing is resolved by name on an executor. Only `increment`/`add` are used. They are drained **once**, at the commit boundary, in `BaseHoodieClient#updateExtraMetadata`. The single drain is load-bearing: both sinks consume the registry destructively — the drain clears it, and `Registry.getAllMetrics(flush=true, …)` clears it when a reporter scrapes — so draining twice would give commit metadata *or* the reporter, never both. One read fans out to the commit's extra metadata as `hoodie.rli.lookup.<caller>.<metric>` and to the metrics reporter as gauges, then clears. Since every engine write path reaches `commitStats`, this covers Spark DataSource, all four Spark SQL DML commands and StreamSync by construction. It also gives the DeltaStreamer path a reporting cadence it did not have: `registerHoodieCommonMetrics` is otherwise reachable only from `Metrics.flush()` (no production callers) and `Metrics.shutdownAllMetrics()` (called only by `HoodieSparkSqlWriter`), so a long-running streaming job previously published these only from the JVM shutdown hook. **`test(metrics)`: functional coverage** on all three write paths, against both the global and the partitioned record index — separate closures on separate code paths. Assertions read counters off the latest commit on the timeline. Invariant throughout: `hits + misses == records_looked_up == incoming keys`. Measured, all passing: | Path | Global RLI | Partitioned RLI | |---|---|---| | DataSource upsert (20 updates + 1 insert) | 20 / 1 / 21, 8 shards | 20 / 1 / 21, 3 shards | | SQL `UPDATE` (optimized writes off) | 60 / 0 / 60, 10 shards | 60 / 0 / 60, 3 shards | | SQL `MERGE INTO` | 25 / 0 / 25 | 25 / 0 / 25 | | DeltaStreamer `sync()` ×2 | 500 / 500 / 1000, 10 shards | 500 / 500 / 1000, 3 shards | ### Impact **New config** `hoodie.metrics.rli.lookup.enable`, default `true`. When off, the registry is never resolved and executors do no extra work. **Behaviour worth flagging for reviewers.** Spark SQL `UPDATE` and `DELETE` perform no record-index lookup at all — they set `_hoodie.spark.sql.writes.prepped=true` whenever `hoodie.spark.sql.optimized.writes.enable` is on (it defaults to `true`), and a prepped write already knows each record's location from the rows it just read. Reporting nothing there is correct rather than a gap, and there are tests pinning both sides so a future change to the prepped path cannot silently alter what operators see. `MERGE INTO` is not a prepped write and does report. **Timeline footprint.** The commit-metadata sink adds roughly eight short entries per commit on an RLI table, and commit metadata is retained permanently including archives. That is the main reason for the config gate, and a fair thing to argue about — see the open question below. **Gauge semantics.** The reporter sink publishes gauges, so each commit overwrites the previous value: the reported number is the last commit's, not a running total. `rate()` will not behave as one might expect on these. No format change, no public API break. ### Risk Level low Additive and gated. The registry changes are the only ones touching existing behaviour: the `set()` rejection has no production callers today, and the staleness eviction only fires where updates were already being silently dropped. Verified by 11 registry tests plus 18 functional tests across three write paths and both index variants. ### Documentation Update The new config `hoodie.metrics.rli.lookup.enable` carries a full description, including the timeline-retention caveat. Happy to add a website page describing the counters if reviewers think this warrants one. ### Open question for reviewers Is `true` the right default? Observability out of the box is the argument for; permanent per-commit timeline growth is the argument against. I lean toward defaulting to `false` for a release or two and flipping once it has baked, and would rather take that steer now than after it ships. ### Contributor's checklist - [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute) - [x] Enough context is provided in the sections above - [x] Adequate tests were added if applicable 🤖 Generated with [Claude Code](https://claude.com/claude-code) https://claude.ai/code/session_01VZsqnoc1EKhEa7459mh6sK -- 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]
