rahil-c commented on code in PR #19575: URL: https://github.com/apache/hudi/pull/19575#discussion_r3859019502
########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/RliLookupMetricsTestBase.scala: ########## @@ -0,0 +1,136 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.functional + +import org.apache.hudi.DataSourceWriteOptions +import org.apache.hudi.common.config.HoodieMetadataConfig +import org.apache.hudi.common.metrics.Registry +import org.apache.hudi.config.HoodieIndexConfig +import org.apache.hudi.metrics.RecordIndexMetricNames + +import scala.collection.JavaConverters._ + +/** + * Shared plumbing for the record level index lookup metric tests: index selection, and reading the counters back the way an operator would -- off the la + */ +abstract class RliLookupMetricsTestBase extends RecordLevelIndexTestBase { Review Comment: Partly taken. Deleted outright: `TestRliLookupMetricsReporting` (reporting is now the only sink, so every remaining test asserts through a reporter), `TestExecutorMetricsGenericity` (the abstraction is gone), and the Spark SQL class (that path does not report -- apache/hudi#19740). Not taking the rest yet. The MOR twins and the multi-table class are the only coverage of the two things most likely to break silently here -- table-type independence and per-table key separation -- and I would rather cut them once the registry lifetime settles, since that is what decides whether the digest is still load-bearing. Your point that different `TBL_NAME`s separate the keys on their own is fair; that makes the multi-table class weaker than intended rather than useless, and it is on the list. ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/ExecutorMetricRegistry.java: ########## @@ -0,0 +1,107 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.metrics; + +import org.apache.hudi.common.metrics.Registry; +import org.apache.hudi.config.HoodieWriteConfig; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.function.Predicate; + +/** + * Every class of executor-collected metric, and the only thing a new one is added to. The driver must + * declare it up front because an {@code AccumulatorV2} must be registered with the {@code SparkContext} + * before a task can contribute; the bundle sent to executors and the commit drain both iterate this. + */ +public enum ExecutorMetricRegistry implements ExecutorMetricGroup { + + RECORD_INDEX_LOOKUP( + "HoodieRecordIndexLookup", + "hoodie.rli.lookup.", + "rli", + "lookup", + HoodieWriteConfig::isRecordIndexLookupMetricsEnabled); + + private final String registryName; + private final String commitMetadataPrefix; + private final String metricAction; + private final String metricQualifier; + private final Predicate<HoodieWriteConfig> enabled; + + ExecutorMetricRegistry(String registryName, String commitMetadataPrefix, String metricAction, + String metricQualifier, Predicate<HoodieWriteConfig> enabled) { + this.registryName = registryName; + this.commitMetadataPrefix = commitMetadataPrefix; + this.metricAction = metricAction; + this.metricQualifier = metricQualifier; + this.enabled = enabled; + } + + /** The bare name emitting code passes to {@link Registry#getRegistry(String)}. */ + @Override + public String registryName() { + return registryName; + } + + @Override + public String commitMetadataPrefix() { + return commitMetadataPrefix; + } + + @Override + public String metricAction() { + return metricAction; + } + + @Override + public String metricQualifier() { + return metricQualifier; + } + + /** Gating here, rather than in the drain, is what lets a new class of metric need no new config. */ + @Override + public boolean isEnabled(HoodieWriteConfig config) { + return enabled.test(config); + } + + /** + * Driver-side {@code REGISTRY_MAP} key. Table name alone is not an identity: two tables can share one + * and would then share a registry. Executors use the bare {@link #registryName()}. + */ + @Override + public String scopedName(String basePath) { Review Comment: Real, and I hit it independently from the other direction while debugging: `Metrics.shutdown()` running `Registry.getAllMetrics(true, true)` is also why an abandoned commit's counters cannot survive to the retry on the DataSource path. In practice the republish does not fire today, because release-by-subtraction leaves the registry empty by the time teardown runs -- so it needs leftovers to be visible at all. Deferring the digest-free name / scrape exclusion to the registry-lifetime follow-up, since the digest only exists to disambiguate inside the process-wide map and both go away together. ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/RecordIndexMetricNames.java: ########## @@ -0,0 +1,56 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.metrics; + +/** + * Counter names for the record index lookup phase. Collection itself is generic -- see + * {@link ExecutorMetricRegistry} and {@link ExecutorMetrics}. + */ +public class RecordIndexMetricNames { + + /** Scoping, prefix and reporter naming all live on the enum entry. */ + public static final String REGISTRY_NAME = ExecutorMetricRegistry.RECORD_INDEX_LOOKUP.registryName(); + + public static final String COMMIT_METADATA_PREFIX = + ExecutorMetricRegistry.RECORD_INDEX_LOOKUP.commitMetadataPrefix(); + + // Counters are tagged by caller so dedupe traffic is distinguishable from tag-location traffic. + public static final String CALLER_TAG_LOCATION = "tag"; + public static final String CALLER_DEDUPE = "dedupe"; + + public static final String KEY_COUNT = "lookup_record_index_key_count"; + public static final String KEY_HIT_COUNT = "lookup_record_index_key_hit_count"; + public static final String KEY_MISS_COUNT = "lookup_record_index_key_miss_count"; + public static final String SHARDS_READ = "lookup_record_index_shards_read"; + /** Wall-clock spent in the shard read, summed across shards. Revives the third dead upstream constant, + * {@code HoodieMetadataMetrics.LOOKUP_RECORD_INDEX_TIME_STR}. */ + public static final String LOOKUP_TIME = "lookup_record_index_time"; Review Comment: Fair, and I have documented the difference rather than dropped it. `index.lookup.duration` is driver wall-clock for the whole `tagLocation` including scheduling; `lookup_record_index_time` is the sum of per-shard executor time inside the index read. Dividing the latter by `shards_read` gives mean per-shard cost, and comparing the two shows how much of the lookup was actually index reading. That said, if you would rather this PR ship four counters and leave timing to the existing metric, I have no objection -- say the word. ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/RecordIndexMetricNames.java: ########## @@ -0,0 +1,56 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.metrics; + +/** + * Counter names for the record index lookup phase. Collection itself is generic -- see + * {@link ExecutorMetricRegistry} and {@link ExecutorMetrics}. + */ +public class RecordIndexMetricNames { + + /** Scoping, prefix and reporter naming all live on the enum entry. */ + public static final String REGISTRY_NAME = ExecutorMetricRegistry.RECORD_INDEX_LOOKUP.registryName(); + + public static final String COMMIT_METADATA_PREFIX = + ExecutorMetricRegistry.RECORD_INDEX_LOOKUP.commitMetadataPrefix(); + + // Counters are tagged by caller so dedupe traffic is distinguishable from tag-location traffic. + public static final String CALLER_TAG_LOCATION = "tag"; + public static final String CALLER_DEDUPE = "dedupe"; + + public static final String KEY_COUNT = "lookup_record_index_key_count"; Review Comment: (1) Taken as a doc fix: the value counts records, not distinct keys. Renaming the constant would change a metric name the requester specified, so the javadoc and config doc now say records explicitly instead. (2) The dead `HoodieMetadataMetrics.LOOKUP_RECORD_INDEX_*` constants and the `TODO [HUDI-9544]` are a good catch. Deferring to the same follow-up rather than touching `HoodieBackedTableMetadata` in a metrics PR, but it should be cleaned up when this lands. ########## hudi-io/src/main/java/org/apache/hudi/common/metrics/Registry.java: ########## @@ -190,6 +199,28 @@ static void setRegistries(Collection<Registry> registries) { */ void set(String name, long value); + /** + * Subtract a set of counts previously read out of this registry, clamping every counter at zero. + * + * Used to hand a batch of counters over to a consumer that owns them from then on -- the commit-boundary + * drain for the record index lookup counters -- without discarding whatever arrived after they were read. + * + * Clamping is what distinguishes this from {@code add(name, -value)}. The registry can be emptied + * underneath a caller by an unrelated destructive scrape ({@link #getAllMetrics(boolean, boolean)} with + * {@code flush=true} clears every registry in the process), and an unbounded subtraction would then leave + * negative counters behind for good. + * + * The default is a best-effort read-modify-write. Implementations able to do this atomically should + * override it, and should drop counters that reach zero rather than leaving them at zero, so a registry + * nobody is writing to reads as empty. + * + * @param counts the counts to release, as returned by {@link #getAllCounts(boolean)}. + */ + default void release(Map<String, Long> counts) { Review Comment: Correct that the default is unreachable in production today -- the only `release` caller is the drain, whose registry is always a `DistributedRegistry`. Leaving it for now because the `LocalRegistry` override is what keeps the two implementations behaving the same way (it evicts counters reaching zero rather than setting them to zero, which the interface default cannot express), and dropping the default would make `release` abstract on a public hudi-io interface. Folding into the same hudi-io follow-up as the `Registry.getRegistry` contract change. ########## hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestDistributedRegistry.java: ########## @@ -184,6 +187,131 @@ public void testAddMetricsParallel() { Assertions.assertEquals(finalExpectedSum, metricCounts.get(METRIC_1)); } + @Test + public void testSetThrowsOnExecutor() { + // Given: a registry registered to the spark context + String registryName = REGISTRY_NAME + "_testSetOnExecutor"; + Registry registry = engineContext.getMetricRegistry("", registryName); + + List<Integer> data = new ArrayList<>(); + data.add(1); + + // When/Then: set() invoked on an executor must fail - it is non-commutative under accumulator merges. + // The UnsupportedOperationException thrown on the executor surfaces wrapped in a SparkException. + assertFailsOnExecutorWith("DistributedRegistry.set() must not be called from a Spark executor", () -> + engineContext.map(data, value -> { + registry.set(METRIC_1, value); + return null; + }, 1)); + } + + @Test + public void testReleaseThrowsOnExecutor() { + // Given: a registry registered to the spark context + String registryName = REGISTRY_NAME + "_testReleaseOnExecutor"; + Registry registry = engineContext.getMetricRegistry("", registryName); + + List<Integer> data = new ArrayList<>(); + data.add(1); + + // When/Then: release() invoked on an executor must fail - clamping and eviction are order-dependent + // under accumulator merges. The UnsupportedOperationException surfaces wrapped in a SparkException. + assertFailsOnExecutorWith("DistributedRegistry.release() must not be called from a Spark executor", () -> + engineContext.map(data, value -> { + registry.release(Collections.singletonMap(METRIC_1, (long) value)); + return null; + }, 1)); + } + + /** + * Asserts the job failed because the executor-side guard fired, not for some unrelated reason such as a serialization error. + */ + private static void assertFailsOnExecutorWith(String expectedMessage, Executable executable) { + SparkException thrown = Assertions.assertThrows(SparkException.class, executable); + StringBuilder chain = new StringBuilder(); + for (Throwable t = thrown; t != null; t = t.getCause()) { + chain.append(t).append('\n'); + if (t.getCause() == t) { + break; + } + } + Assertions.assertTrue(chain.toString().contains(expectedMessage), + "expected the executor-side guard to fail the job, got: " + chain); + } + + @Test + public void testSetOnDriverSucceeds() { + // set() on the driver (no TaskContext) remains supported. + DistributedRegistry registry = new DistributedRegistry(REGISTRY_NAME + "_testSetOnDriver"); + registry.set(METRIC_1, 42); + Assertions.assertEquals(42, registry.getAllCounts().get(METRIC_1)); + } + + @Test + public void testGetMetricRegistryReplacesNonDistributedRegistry() { Review Comment: You are right that the test never enters the branch it names -- `getMetricRegistry` removes the seeded `LocalRegistry` before `getRegistryOfClass` runs, so `computeIfAbsent` creates a fresh `DistributedRegistry` and the `!(instanceof)` fallback stays uncovered. Deferring with the branch itself: if the registry becomes instance-scoped, the fallback goes away and so does the test. Noting it on the follow-up so it is not silently lost. -- 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]
