voonhous commented on code in PR #18942:
URL: https://github.com/apache/hudi/pull/18942#discussion_r3829506189
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java:
##########
@@ -456,8 +459,12 @@ public void updateClusteringFileCreationMetrics(long
durationInMs) {
reportMetrics(HoodieTimeline.CLUSTERING_ACTION, "fileCreationTime",
durationInMs);
}
- public void updateTableServiceInstantMetrics(final HoodieActiveTimeline
activeTimeline) {
- updateEarliestPendingInstant(activeTimeline,
EARLIEST_PENDING_CLUSTERING_INSTANT_STR, HoodieTimeline.CLUSTERING_ACTION);
+ public void updateTableServiceInstantMetrics(final HoodieActiveTimeline
activeTimeline, final InstantGenerator instantGenerator) {
Review Comment:
Storage IO on every commit even when metrics are off, which is the default
(`hoodie.metrics.on`, `HoodieMetricsConfig.java:58-62`). The only guard is
inside `updateMetric` (`:583`), i.e. after the reads. Pre-PR this method was
pure in-memory matching, so that was free.
It is also the one entry point `1fd2c3671a80` (#18947) missed:
`testUpdateMethodsAreNoOpsWhenMetricsOff` (`TestHoodieMetrics.java:411`) lists
19 update methods, not this one.
Please add the guard, and add this method to that test.
```suggestion
public void updateTableServiceInstantMetrics(final HoodieActiveTimeline
activeTimeline, final InstantGenerator instantGenerator) {
if (!config.isMetricsOn()) {
return;
}
```
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java:
##########
@@ -456,8 +459,12 @@ public void updateClusteringFileCreationMetrics(long
durationInMs) {
reportMetrics(HoodieTimeline.CLUSTERING_ACTION, "fileCreationTime",
durationInMs);
}
- public void updateTableServiceInstantMetrics(final HoodieActiveTimeline
activeTimeline) {
- updateEarliestPendingInstant(activeTimeline,
EARLIEST_PENDING_CLUSTERING_INSTANT_STR, HoodieTimeline.CLUSTERING_ACTION);
+ public void updateTableServiceInstantMetrics(final HoodieActiveTimeline
activeTimeline, final InstantGenerator instantGenerator) {
+ // Clustering is scheduled as CLUSTERING_ACTION only on timeline layout 2.
On table version six it is scheduled as
+ // REPLACE_COMMIT_ACTION, which insert_overwrite and delete_partition
share, so the clustering plan is what
+ // identifies it rather than the action name.
+ Predicate<HoodieInstant> pendingClustering = instant ->
isPendingClusteringInstant(activeTimeline, instant, instantGenerator);
+ updateEarliestPendingInstant(activeTimeline,
EARLIEST_PENDING_CLUSTERING_INSTANT_STR, HoodieTimeline.CLUSTERING_ACTION,
pendingClustering);
Review Comment:
`HoodieTimeline.filterPendingClusteringTimeline()` already does this, per
layout: `BaseTimelineV1.java:62-68` is this exact predicate,
`BaseTimelineV2.java:67-71` is action-name-only with zero IO. Hand-rolled,
tv8/9/10 pays a plan read per pending insert_overwrite / delete_partition on
every commit to learn what the layout already guarantees.
It also runs twice: `filter()` is eager (`BaseHoodieTimeline.java:96,358`),
so `firstInstant()` (`:502`) does not short-circuit and `:578` filters again.
2N `storage.open()` per commit.
Please call `activeTimeline.filterPendingClusteringTimeline()` once and feed
both metrics from it. That drops the `Predicate` overloads, the
`InstantGenerator` param and both `BaseHoodieWriteClient` hunks.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java:
##########
@@ -456,8 +459,12 @@ public void updateClusteringFileCreationMetrics(long
durationInMs) {
reportMetrics(HoodieTimeline.CLUSTERING_ACTION, "fileCreationTime",
durationInMs);
}
- public void updateTableServiceInstantMetrics(final HoodieActiveTimeline
activeTimeline) {
- updateEarliestPendingInstant(activeTimeline,
EARLIEST_PENDING_CLUSTERING_INSTANT_STR, HoodieTimeline.CLUSTERING_ACTION);
+ public void updateTableServiceInstantMetrics(final HoodieActiveTimeline
activeTimeline, final InstantGenerator instantGenerator) {
+ // Clustering is scheduled as CLUSTERING_ACTION only on timeline layout 2.
On table version six it is scheduled as
+ // REPLACE_COMMIT_ACTION, which insert_overwrite and delete_partition
share, so the clustering plan is what
+ // identifies it rather than the action name.
Review Comment:
nit, feel free to ignore: wrong field. Selection is on `operationType`
(`ClusteringUtils.java:285`), not the plan -- both test fixtures carry a plan.
Same wording in the commit message and PR body.
```suggestion
// Clustering is scheduled as CLUSTERING_ACTION only on timeline layout
2. On table version six it is scheduled as
// REPLACE_COMMIT_ACTION, which insert_overwrite and delete_partition
share, so the requested replace metadata's
// operationType is what identifies it rather than the action name.
```
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java:
##########
@@ -484,13 +491,35 @@ private void updateEarliestPendingInstant(final
HoodieActiveTimeline activeTimel
final String metricName,
final String action) {
Set<String> validActions = CollectionUtils.createSet(action);
- HoodieTimeline filteredInstants =
activeTimeline.filterInflightsAndRequested().filter(instant ->
validActions.contains(instant.getAction()));
+ updateEarliestPendingInstant(activeTimeline, metricName, action, instant
-> validActions.contains(instant.getAction()));
+ }
+
+ private void updateEarliestPendingInstant(final HoodieActiveTimeline
activeTimeline,
+ final String metricName,
+ final String action,
+ final Predicate<HoodieInstant>
pendingFilter) {
+ HoodieTimeline filteredInstants =
activeTimeline.filterInflightsAndRequested().filter(pendingFilter);
Option<HoodieInstant> hoodieInstantOption =
filteredInstants.firstInstant();
if (hoodieInstantOption.isPresent()) {
updateTimestampMetric(metricName, action, hoodieInstantOption);
}
}
+ /**
+ * Whether a pending instant belongs to a clustering operation, on any table
version.
+ * A plan that cannot be read is treated as non-clustering so that metrics
never fail the commit.
+ */
+ private boolean isPendingClusteringInstant(final HoodieActiveTimeline
activeTimeline,
+ final HoodieInstant instant,
+ final InstantGenerator
instantGenerator) {
+ try {
+ return ClusteringUtils.isClusteringInstant(activeTimeline, instant,
instantGenerator);
+ } catch (Exception e) {
+ log.warn("Failed to read the replace metadata of {} while updating
clustering instant metrics", instant, e);
+ return false;
+ }
Review Comment:
The catch is right and worth keeping: `postCommit` has no `catch`, and
`hoodie.write.can.ignore.post.commit.failures` defaults to `false`
(`HoodieWriteConfig.java:747-748`), so an escape would fail the commit.
Two asks. Nothing tests it -- add a throwing read asserting no throw and
count 0, otherwise the fallback degrades silently to the exact bug being fixed.
And log at `debug`, or drop the exception arg: `isClusteringInstant` uses the
3-arg `getClusteringPlan`, which opts out of #18288's rollback recovery, so a
concurrently rolled-back replacecommit logs a stacktrace per instant per commit.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metrics/TestHoodieMetrics.java:
##########
@@ -320,13 +324,61 @@ public void testTimerCtxandGauges() throws
InterruptedException {
assertEquals((long)metrics.getRegistry().getGauges().get(metricName).getValue(),
6L);
}
+ @Test
+ void testPendingClusteringInstantMetricsOnTableVersionSix() {
Review Comment:
Nothing in this test is table version six. `MockHoodieActiveTimeline extends
ActiveTimelineV2` (`:349`) and `INSTANT_GENERATOR` is `DefaultInstantGenerator
extends InstantGeneratorV2`; tv6 is `LAYOUT_VERSION_1`
(`HoodieTableVersion.java:54`), so it gets `ActiveTimelineV1` +
`InstantGeneratorV1`. The `readRequestedReplaceMetadata` override also stubs
the only method touching storage -- it has to, since the no-arg
`ActiveTimelineV2()` leaves `metaClient` null (`:88`).
`TestBaseHoodieTableServiceClient` in this module already has the harness:
`initMetaClient(preTableVersion8)` gives `HoodieTableVersion.SIX`,
`createClusteringInstant(time, REPLACE_COMMIT_ACTION)` (`:429-449`) writes a
real pending replacecommit clustering instant, pattern at `:299-315`.
Please rebuild on that, or rename so it does not claim tv6 coverage.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metrics/TestHoodieMetrics.java:
##########
@@ -320,13 +324,61 @@ public void testTimerCtxandGauges() throws
InterruptedException {
assertEquals((long)metrics.getRegistry().getGauges().get(metricName).getValue(),
6L);
}
+ @Test
+ void testPendingClusteringInstantMetricsOnTableVersionSix() {
+ // Table version six schedules clustering as REPLACE_COMMIT_ACTION, which
insert_overwrite shares.
+ HoodieInstant pendingInsertOverwrite =
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION, "1001");
+ HoodieInstant pendingClustering =
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION, "1002");
+ Map<String, String> operationTypes = new HashMap<>();
+ operationTypes.put("1001", WriteOperationType.INSERT_OVERWRITE.name());
+ operationTypes.put("1002", WriteOperationType.CLUSTER.name());
Review Comment:
Two gaps.
No INFLIGHT instant. `filterInflightsAndRequested()` returns both states,
and inflight takes a different branch (`ClusteringUtils.java:182-186`, the
reason for #2389 / HUDI-1498). Add one in `State.INFLIGHT` and assert the count
becomes 2.
`"INSERT_OVERWRITE"` is not a shape production writes.
`createRequestedCommitWithReplaceMetadata` writes a bare `new
HoodieRequestedReplaceMetadata()` (`ActiveTimelineV2.java:143`), so real
insert_overwrite has `operationType == null`. Add a null case --
`DELETE_PARTITION` is the one that is genuinely non-null.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metrics/TestHoodieMetrics.java:
##########
@@ -320,13 +324,61 @@ public void testTimerCtxandGauges() throws
InterruptedException {
assertEquals((long)metrics.getRegistry().getGauges().get(metricName).getValue(),
6L);
}
+ @Test
+ void testPendingClusteringInstantMetricsOnTableVersionSix() {
+ // Table version six schedules clustering as REPLACE_COMMIT_ACTION, which
insert_overwrite shares.
+ HoodieInstant pendingInsertOverwrite =
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION, "1001");
+ HoodieInstant pendingClustering =
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION, "1002");
+ Map<String, String> operationTypes = new HashMap<>();
+ operationTypes.put("1001", WriteOperationType.INSERT_OVERWRITE.name());
+ operationTypes.put("1002", WriteOperationType.CLUSTER.name());
+
+ hoodieMetrics.updateTableServiceInstantMetrics(
+ new MockClusteringPlanTimeline(operationTypes, pendingInsertOverwrite,
pendingClustering), INSTANT_GENERATOR);
+
+ // Only the instant carrying a clustering plan counts. insert_overwrite is
the earlier of the two, so
+ // both assertions fail if the action name alone is used to select it.
Review Comment:
nit, feel free to ignore: same wrong field -- both fixtures carry a plan,
`operationType` is what separates them.
```suggestion
// Only the instant whose requested replace metadata carries
operationType CLUSTER counts. Both fixtures
// deliberately carry a plan, so this also pins that selection must not
key on plan presence.
// insert_overwrite is the earlier of the two, so both assertions fail
if the action name alone selects it.
```
Also both assertions deref the gauge with no null check: on a regression it
is never registered (`HoodieMetrics.java:501-505`) and line 346 NPEs instead of
failing readably.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metrics/TestHoodieMetrics.java:
##########
@@ -320,13 +324,61 @@ public void testTimerCtxandGauges() throws
InterruptedException {
assertEquals((long)metrics.getRegistry().getGauges().get(metricName).getValue(),
6L);
}
+ @Test
+ void testPendingClusteringInstantMetricsOnTableVersionSix() {
+ // Table version six schedules clustering as REPLACE_COMMIT_ACTION, which
insert_overwrite shares.
+ HoodieInstant pendingInsertOverwrite =
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION, "1001");
+ HoodieInstant pendingClustering =
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION, "1002");
+ Map<String, String> operationTypes = new HashMap<>();
+ operationTypes.put("1001", WriteOperationType.INSERT_OVERWRITE.name());
+ operationTypes.put("1002", WriteOperationType.CLUSTER.name());
+
+ hoodieMetrics.updateTableServiceInstantMetrics(
+ new MockClusteringPlanTimeline(operationTypes, pendingInsertOverwrite,
pendingClustering), INSTANT_GENERATOR);
+
+ // Only the instant carrying a clustering plan counts. insert_overwrite is
the earlier of the two, so
+ // both assertions fail if the action name alone is used to select it.
+ String countMetric =
hoodieMetrics.getMetricsName(HoodieTimeline.CLUSTERING_ACTION,
HoodieMetrics.PENDING_CLUSTERING_INSTANT_COUNT_STR);
+ assertEquals(1L, (long)
metrics.getRegistry().getGauges().get(countMetric).getValue());
+ String earliestMetric =
hoodieMetrics.getMetricsName(HoodieTimeline.CLUSTERING_ACTION,
HoodieMetrics.EARLIEST_PENDING_CLUSTERING_INSTANT_STR);
+ assertEquals(1002L, (long)
metrics.getRegistry().getGauges().get(earliestMetric).getValue());
+ }
+
private static class MockHoodieActiveTimeline extends ActiveTimelineV2 {
public MockHoodieActiveTimeline(HoodieInstant... instants) {
super();
this.setInstants(Arrays.asList(instants));
}
}
+ /**
+ * Serves a requested replace metadata per instant time so pending
replacecommits can be told apart
+ * by their write operation type without a backing timeline on storage.
+ */
+ private static class MockClusteringPlanTimeline extends
MockHoodieActiveTimeline {
Review Comment:
nit, feel free to ignore: this builder chain is byte-for-byte
`TestBaseHoodieTableServiceClient.java:434-441` in this same module. Moving the
test onto that harness removes it; otherwise consider a shared helper rather
than a third copy.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:
##########
@@ -633,7 +633,7 @@ private void completeClustering(HoodieReplaceCommitMetadata
replaceCommitMetadat
if (clusteringTimer != null) {
long durationInMs = metrics.getDurationInMs(clusteringTimer.stop());
TimelineUtils.parseDateFromInstantTimeSafely(clusteringCommitTime).ifPresent(parsedInstant
->
- metrics.updateCommitMetrics(parsedInstant.getTime(), durationInMs,
replaceCommitMetadata, HoodieActiveTimeline.CLUSTERING_ACTION)
+ metrics.updateCommitMetrics(parsedInstant.getTime(), durationInMs,
replaceCommitMetadata, clusteringInstant.getAction())
Review Comment:
Closing this out -- the convention ended up settled the other way.
Metrics in this class are named after the table service, not the completed
action: compaction reports under `compaction.*` though it completes as
`commit`, log compaction under `logcompaction.*` though it completes as
`deltacommit`. `replacecommit.*` would also share gauges with insert_overwrite
and delete_partition under last-write-wins in `Metrics.registerGauge`. So
master was already correct, and both files are byte-identical to master on this
branch now.
@nsivabalan reopen if you still want `replacecommit` -- that is its own PR
plus a release note, since `clustering.timer` and `clustering.fileCreationTime`
would have to move with it.
--
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]
