feat(metrics): report record index lookup counters at the commit boundary - #19575
Open
rahil-c wants to merge 7 commits into
Open
feat(metrics): report record index lookup counters at the commit boundary#19575rahil-c wants to merge 7 commits into
rahil-c wants to merge 7 commits into
Conversation
…ross-client leaks DistributedRegistry instances are cached in process-wide static maps that outlive the SparkContext, the write client and the table. Three consequences, all on the path any executor-side metric has to travel: - 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() yields a non-deterministic value. It is now rejected from inside a task; increment()/add() remain available. No production caller exists today. - After a SparkContext restart in the same JVM (shells, notebooks, Spark Connect), the cached registry stays bound to the dead context and executor updates silently stop 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, because AccumulatorV2.register() throws if the accumulator already carries registration metadata. Evict-and-recreate is done atomically, or racing callers leave two live accumulators for one metric name with only one of them reachable. - HoodieWrapperFileSystem counters leaked into the next write client for the same table. They are now cleared both when the registries are created and when the client closes. Relates to apache#19063.
…dary HoodieMetadataMetrics has declared lookup_record_index_time, lookup_record_index_key_count and lookup_record_index_key_hit_count for some time, but nothing references them, and readRecordIndexLocationsWithKeys carries "TODO [HUDI-9544]: Metric does not work for rdd based API due to lazy evaluation." The numbers exist -- keysToLookup.size() and the size of the returned map are local variables inside RecordIndexFileGroupLookupFunction.call() -- but the function returns only the hits, so a miss produces no output row and the driver cannot recover the denominator from the resulting RDD at any price. Executors could always count; they had no way to report. This adds four counters per RLI shard (records_looked_up, hits, misses, shards_read), 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 travels with the closure and nothing is looked up by name on the executor. Only increment/add are used. They are drained once, at the commit boundary, in BaseHoodieClient#updateExtraMetadata. A single drain matters: 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 yield 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. Because every engine write path reaches commitStats -- Spark DataSource, all four Spark SQL DML commands and StreamSync -- this covers them all 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 streaming job previously published metrics only from the JVM shutdown hook. Gated by hoodie.metrics.rli.lookup.enable, default true. The gate matters because commit metadata is retained in the timeline permanently including archives; when it is off the registry is never resolved and executors do no extra work. Note the counters are published as gauges, so each commit overwrites the previous value: the reported number is the last commit's, not a running total. Relates to HUDI-9544 and apache#19063.
…e path Functional coverage across Spark DataSource, Spark SQL and DeltaStreamer, each against both the global and the partitioned record index -- they are separate closures on separate code paths. Assertions read the counters off the latest commit on the timeline, which is what an operator actually consumes. The invariant throughout is hits + misses == records_looked_up == incoming keys. Two behaviours worth calling out, both pinned by tests: - 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, so there are tests for both sides: empty with optimized writes on, exact counts with it off. MERGE INTO is not a prepped write and does report. - TestRliLookupMetricsReporting asserts commit metadata and a live ConsoleMetricsReporter are both populated from the same write. That is precisely what a double drain would break, since either sink clears the registry as it reads. RliLookupMetricsTestBase is a class rather than a trait deliberately: the helpers touch metaClient, a protected Java field on HoodieCommonTestHarness, and a Scala trait compiles to a separate class that cannot legally reach it.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19575 +/- ##
============================================
- Coverage 77.09% 75.76% -1.33%
+ Complexity 32490 32111 -379
============================================
Files 2522 2524 +2
Lines 139112 139267 +155
Branches 16714 16768 +54
============================================
- Hits 107243 105511 -1732
- Misses 24291 26102 +1811
- Partials 7578 7654 +76
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
…ters Nine issues found reviewing the previous three commits. Correctness: - The drain was hooked into BaseHoodieClient#updateExtraMetadata, which BaseHoodieTableServiceClient also calls when scheduling table services. Scheduling a compaction or clustering therefore consumed the counters and stamped them into the *plan*, so the next real commit published nothing. Moved to commitStats, on the commit path only. - The drain also ran before the commit was written, so a conflict in preCommit or a failure in commit destroyed the counters and published reporter gauges for work that was rolled back. Split into snapshotIntoCommitMetadata (before, non-consuming) and publishAndRelease (after the commit lands). - Releasing now subtracts exactly what was published instead of clearing the registry. A clear also discarded anything that arrived between the snapshot and the release -- a straggler task's accumulator update, or a concurrent lookup stage. - resolveRegistry tested `instanceof Serializable`, which is always true: Registry itself extends Serializable. The guard never fired, so with a non-Spark engine context a driver-only LocalRegistry would serialize into the closure and collect into per-executor copies nothing reads, reporting zero. Tests for DistributedRegistry instead, which is the only implementation that aggregates back to the driver. - misses was keysLookedUp - hits, mixing a raw input count with a deduplicated hit count, so a batch containing the same key twice reported a phantom miss. Counts distinct keys now. Only differs when hoodie.combine.before.upsert is turned off. - DistributedRegistry#register stamped registeredAppId even when it skipped registration, letting any caller re-brand an accumulator bound to a dead SparkContext with the current application id -- masking the staleness the field exists to detect. - getMetricRegistry could hand back a non-distributed registry: another thread can insert one under the shared key between the remove and the create, and getRegistryOfClass only logs the mismatch. It now replaces the entry rather than returning one that silently collects nothing. Scope and hygiene: - Dropped the HoodieWrapperFileSystem registry clearing entirely. It is unrelated to the record index counters, and clearing on write-client close ran *before* Metrics.shutdownAllMetrics reported, so every existing <table>.HoodieWrapperFileSystem.* gauge would have silently disappeared for users on the DataSource path. - Restored the one-argument PartitionedRecordIndexFileGroupLookupFunction constructor rather than breaking a public signature and passing two nulls at the read call site. - setCaller returns the previous label and restoreCaller puts it back, so nested tagging cannot silently reset the label to tag-location. - sinceVersion corrected to 1.3.0, matching the tree. - TestDistributedRegistryLifecycle removes its entries from both process-wide registry maps, instead of leaving accumulators bound to stopped SparkContexts for whatever runs next in the JVM.
rahil-c
force-pushed
the
rli-lookup-metrics
branch
from
August 11, 2026 23:56
c4fa5fa to
5d3fa9e
Compare
rahil-c
marked this pull request as ready for review
August 12, 2026 00:30
…ecompute The record index lookup counters are incremented inside a Spark transformation, which carries no exactly-once guarantee. That was documented as at-least-once but never measured, and a reviewer on apache#19063 asked specifically for it to be verified. Both halves are now pinned, using their own SparkContext because the shared harness runs local[*], which never retries -- local[n, maxFailures] is required for a failed task to be re-attempted at all. A failed attempt contributes nothing. Partition 0 throws on its first attempt and succeeds on the retry; 100 rows are processed and 100 are counted, because Spark ships accumulator updates home only from attempts that succeed. This is the guarantee that makes ordinary task retries safe for these counters. A repeated evaluation does double count. Two actions over the same uncached RDD yield 200 for a single pass of 100. This is the shape of the real exposure -- a duplicate successful attempt under speculation, or a stage recomputed after a fetch failure, reaches the accumulator the same way. Asserting it keeps the caveat grounded, and makes a change in either direction visible rather than silent. If exactness is ever required, per-shard counters merged with max would make the increment idempotent.
Two coverage gaps that production users would hit immediately. Merge On Read. Tagging happens in the index, above the layer that decides whether an update rewrites a base file or appends to a log file, so MOR is expected to produce identical counters. Expected-to-be-identical is exactly the kind of claim that quietly stops being true, so the DataSource suite now runs against both table types across both index variants. Measured identical to COW. Multiple tables in one JVM. The registry lives in Registry.REGISTRY_MAP, a process-wide static keyed by table name, and that same static map is the origin of several bugs catalogued in apache#19063 including counters leaking between write clients. The keying is what provides isolation, so it is asserted rather than trusted: two tables are written with interleaved upserts of deliberately different sizes, and each commit must carry only its own counts. A leak shows up as a wrong number rather than only as a missing reset. A second test covers the sequential case, where a table written after another finished must start from zero. Adds a tableTypeOpt hook to the shared test base so the table type can be varied by subclass, matching how the index variant is already selected.
hudi-agent
reviewed
Aug 12, 2026
hudi-agent
left a comment
hudi-agent
left a comment
Contributor
There was a problem hiding this comment.
test. A couple of naming and design nits worth considering.
| * | ||
| * <p>Lives in hudi-client-common rather than hudi-spark-client so the shared commit path can reach it. | ||
| * Because the drain happens on that shared path, it applies to every engine write path -- Spark | ||
| * DataSource, Spark SQL and DeltaStreamer -- without any of them knowing about it. |
Contributor
There was a problem hiding this comment.
🤖 nit: the name RecordIndexMetricNames signals a constants class, but snapshotIntoCommitMetadata and publishAndRelease are substantive operations. Could you consider something like RecordIndexLookupMetricsHelper or RecordIndexCommitDrain — or splitting the constants into a separate RecordIndexMetricNames interface and putting the drain logic in a class with a name that reflects its responsibility?
| * @param commitMetadata the extra-metadata map being assembled for this commit | ||
| * @param config the write config, for the table name and the feature gate | ||
| * @return the snapshotted counters, to be handed to {@link #publishAndRelease} on success; empty | ||
| * when the feature is off or nothing was recorded |
Contributor
There was a problem hiding this comment.
🤖 nit: snapshotIntoCommitMetadata mutates commitMetadata in-place AND returns a separate Map<String, Long> — two output channels from one call. Have you considered returning a small result object (or a Pair) that holds both the snapshot and whatever the caller needs to pass to publishAndRelease, so the side-effectful mutation is not implicit in the signature?
…shard size recordShardLookup built a HashSet over every key routed to the shard, purely to count distinct keys. That allocation grows with shard size: harmless on a small table, but on a large one a shard can hold millions of keys and the set is a second copy of all of them, per task. This is the only allocation the counters added that scaled with data volume. The membership set is now built from the keys that were *found* instead. Both collections are already materialised by the caller, but the found set is bounded by the hit count rather than by shard size, and the partitioned path gets it for free -- its result is already a map keyed by record key, so keySet() costs nothing. Counting switches from distinct keys to records, which is also the better answer to the question being asked. "How many records were read" and "what is the hit rate" are about records, so records_looked_up is now how many incoming records the shard was asked about and hits is how many of them matched. hits + misses == records_looked_up stays exact whether or not the batch contains duplicate keys, and a duplicate no longer shows up as a phantom miss. With hoodie.combine.before.upsert at its default the batch is already deduplicated, so the reported numbers are unchanged for the common configuration. Verified across the full suite: 13 client tests, 24 Spark functional tests spanning DataSource, Spark SQL and both table types against both index variants, and 2 real DeltaStreamer syncs. Also re-checked on a 7.5M row table where a single upsert tags 1.49M keys, which reports exactly 745,654 hits and 745,654 misses.
Collaborator
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Describe the issue this Pull Request addresses
HoodieMetadataMetricshas declared three record-index lookup metrics for some time:Nothing in the repository references any of them, and
HoodieBackedTableMetadata#readRecordIndexLocationsWithKeysrecords why:// 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 insideRecordIndexFileGroupLookupFunction.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. Recoveringrecords_looked_uporshards_readwould 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): hardenDistributedRegistry— 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 anAccumulatorV2the driver merges executor copies in an unspecified order, so an executor-sideset()is non-deterministic. Now rejected from inside a task. No production caller exists today.isRegistered()cannot detect this — it consults a Spark-global weak-reference table thatSparkContext.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, sinceAccumulatorV2.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.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. Onlyincrement/addare used.They are drained once, at the commit boundary, in
BaseHoodieWriteClient#commitStats. The single drain is load-bearing: both sinks consume the registry destructively — andRegistry.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 ashoodie.rli.lookup.<caller>.<metric>and to the metrics reporter as gauges.Two details matter for correctness. The snapshot is taken before the commit is written but the counters are only released after it lands, so a conflict in
preCommitor a failure incommitdoes not destroy them or publish gauges for rolled-back work. And releasing subtracts exactly what was published rather than clearing, so a straggler task's accumulator update arriving mid-drain is carried into the next commit instead of being silently dropped.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:registerHoodieCommonMetricsis otherwise reachable only fromMetrics.flush()(no production callers) andMetrics.shutdownAllMetrics()(called only byHoodieSparkSqlWriter), 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:
UPDATE(optimized writes off)MERGE INTOsync()×2Merge On Read reports identically to Copy On Write (20 / 1 / 21), which is expected — tagging happens in the index, above the layer that decides base-file-rewrite versus log-append — but is asserted rather than assumed.
test(metrics): task retry and stage recompute. Previously documented as at-least-once without being measured; now both halves are pinned, using a dedicated SparkContext because the shared harness runslocal[*], which never retries. A task that fails and is re-attempted contributes nothing from the failed attempt — 100 rows processed, 100 counted. A second evaluation of the same uncached RDD counts twice — 200 for a single pass of 100, which is the shape of the real exposure under speculation or a recomputed stage.test(metrics): multiple tables in one JVM. The registry lives inRegistry.REGISTRY_MAP, a process-wide static keyed by table name, and that same map is the origin of several bugs in #19063 including counters leaking between write clients. Two tables are written with interleaved upserts of different sizes; each commit must carry only its own counts, so a leak surfaces as a wrong number rather than only as a missing reset. Measured: A reports 12, B reports 7, despite A's 30 landing between them.perf(metrics): bounded allocation. An earlier revision built aHashSetover every key routed to a shard in order to count distinct keys — an allocation that grows with shard size, which on a large table means a second copy of millions of keys per task. The membership set is now built from the keys that were found instead, bounding it by hit count; the partitioned path gets it free, since its result is already a map keyed by record key. Counting also switched from distinct keys to records, which is the better answer to "how many records were read" and keepshits + misses == records_looked_upexact regardless of duplicates.Scale check
Run outside CI on a 7.5M row table where one upsert tags 1.49M keys across 10 shards:
Exact. On overhead I can only report that it was not separable from noise on this harness: the bulk-insert phase, which performs no tagging and therefore cannot be affected by this change, varied more between runs than the instrumented upsert did. I would not claim a percentage from it. The harness is not included in this PR — happy to share it if a reviewer wants to reproduce.
Impact
New config
hoodie.metrics.rli.lookup.enable, defaulttrue. When off, the registry is never resolved and executors do no extra work.Behaviour worth flagging for reviewers. Spark SQL
UPDATEandDELETEperform no record-index lookup at all — they set_hoodie.spark.sql.writes.prepped=truewheneverhoodie.spark.sql.optimized.writes.enableis on (it defaults totrue), 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 INTOis 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 39 tests: 13 in hudi-spark-client covering registry lifecycle, SparkContext restart and task-retry semantics; 24 functional tests spanning Spark DataSource, Spark SQL and both table types against both index variants, plus multi-table isolation; and 2 real DeltaStreamer syncs. Plus the out-of-CI scale check above.
Documentation Update
The new config
hoodie.metrics.rli.lookup.enablecarries 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
truethe right default? Observability out of the box is the argument for; permanent per-commit timeline growth is the argument against. I lean toward defaulting tofalsefor a release or two and flipping once it has baked, and would rather take that steer now than after it ships.Contributor's checklist
🤖 Generated with Claude Code
https://claude.ai/code/session_01VZsqnoc1EKhEa7459mh6sK