rahil-c commented on code in PR #19575: URL: https://github.com/apache/hudi/pull/19575#discussion_r3859017918
########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/ExecutorMetrics.java: ########## @@ -0,0 +1,135 @@ +/* + * 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.common.util.Option; +import org.apache.hudi.config.HoodieWriteConfig; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Commit-boundary drain for executor-collected metrics, generic over {@link ExecutorMetricRegistry}. On + * the shared commit path, so it covers Spark DataSource, Spark SQL and DeltaStreamer alike. + */ +public class ExecutorMetrics { + + private ExecutorMetrics() { + } + + /** + * Snapshots into commit metadata without consuming. Split from {@link #publishAndRelease} so a commit + * that never lands neither loses its counters nor publishes gauges for rolled-back work. An all-zero + * registry is skipped to keep residue off the timeline; zeros are otherwise kept, since an explicit + * {@code misses=0} is meaningful. + */ + public static DrainedCounters snapshotIntoCommitMetadata(Map<String, String> commitMetadata, + HoodieWriteConfig config) { + return snapshotIntoCommitMetadata(commitMetadata, config, Arrays.asList(ExecutorMetricRegistry.values())); + } + + /** Visible for testing the collection machinery against a group it does not ship with. */ + static DrainedCounters snapshotIntoCommitMetadata(Map<String, String> commitMetadata, + HoodieWriteConfig config, + Collection<? extends ExecutorMetricGroup> groups) { + List<Drained> drained = new ArrayList<>(); + for (ExecutorMetricGroup metricRegistry : groups) { + if (!metricRegistry.isEnabled(config)) { + continue; + } + Registry registry = Registry.REGISTRY_MAP.get( + Registry.makeKey(config.getTableName(), metricRegistry.scopedName(config.getBasePath()))); + if (registry == null) { + continue; + } + Map<String, Long> counts = new HashMap<>(); + boolean recordedSomething = false; + for (Map.Entry<String, Long> counter : registry.getAllCounts(false).entrySet()) { + if (counter.getValue() == null) { + continue; + } + counts.put(counter.getKey(), counter.getValue()); + recordedSomething |= counter.getValue() != 0L; + } + if (!recordedSomething) { + continue; + } + counts.forEach((name, value) -> + commitMetadata.put(metricRegistry.commitMetadataPrefix() + name, String.valueOf(value))); + drained.add(new Drained(metricRegistry, registry, counts)); + } + return drained.isEmpty() ? DrainedCounters.EMPTY : new DrainedCounters(drained); + } + + /** + * Release subtracts what was published rather than clearing, so a straggler task's update arriving after + * the snapshot survives. Publishing here rather than letting the reporter scrape is what lets both sinks + * work at once: {@link Registry#getAllMetrics} consumes the registry when it scrapes. + */ + public static void publishAndRelease(DrainedCounters counters, HoodieMetrics hoodieMetrics) { Review Comment: Agreed on the direction, reached from the other end. Publishing now happens only after the commit lands, so an abandoned attempt publishes nothing, and the `AcrossFailedCommit` assertion is flipped to "the retry reports only its own lookups" as you suggested. Your parenthetical turned out to be the operative fact: with `hoodie.metrics.on=true` the carry-over does not hold anyway, because `Metrics.shutdown()` runs `Registry.getAllMetrics(true, true)` and clears every registry after each DataSource write. So on that path an abandoned attempt's counters are dropped rather than carried, which is the safer of the two failure modes. That is now stated in the PR description and asserted by the test rather than left implicit. ########## hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDReadClient.java: ########## @@ -210,8 +212,16 @@ public JavaRDD<HoodieRecord<T>> filterExists(JavaRDD<HoodieRecord<T>> hoodieReco * @return Tagged RDD of Hoodie records */ public JavaRDD<HoodieRecord<T>> tagLocation(JavaRDD<HoodieRecord<T>> hoodieRecords) throws HoodieIndexException { - return HoodieJavaRDD.getJavaRDD( - index.tagLocation(HoodieJavaRDD.of(hoodieRecords), context, hoodieTable)); + // Lookups driven from the read client are dedupe traffic, not tag-location traffic. Label them so + // the two are distinguishable in the reported counters. Driver-side only: the label is + // captured when the lookup closure is built. + String previousCaller = RecordIndexLookupMetrics.setCaller(RecordIndexMetricNames.CALLER_DEDUPE); Review Comment: Resolved by removing caller tagging from this PR entirely. You were right that it had no coverage, and the `checkExists` mislabelling you spotted confirms it was not carrying its weight as written. Tracked for restoration with the coverage it was missing, and the first question there is whether the dedupe path is reachable enough to be worth it -- `SparkRDDReadClient.tagLocation` has one production caller and insert-dedupe is off by default. ########## 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.", Review Comment: Moot now -- the counters no longer reach the timeline, so there is no permanent extra-metadata key to name. The reporter gauge names are `<table>.rli.lookup.<metric>`, which follows the existing `<table>.<action>.<metric>` convention used by `HoodieMetrics`. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestRliLookupMetricsOnSparkSql.scala: ########## @@ -0,0 +1,134 @@ +/* + * 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.config.HoodieIndexConfig +import org.apache.hudi.metrics.RecordIndexMetricNames + +import org.apache.spark.sql.SaveMode +import org.junit.jupiter.api.{Tag, Test} +import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue} + +/** Record level index lookup counters on the Spark SQL write path. */ +@Tag("functional") +class TestRliLookupMetricsOnSparkSql extends RliLookupMetricsTestBase { + + private val sqlTable = "rli_lookup_metrics_tbl" + private val numSeedRecords = 60 + + /** + * Seeds a table through the DataSource so the record index exists, then exposes it to SQL and applies + * the index settings as session configs -- index type is a write config, not a table property. + */ + private def seedTableAndRegisterForSql(): Unit = { + doWriteAndValidateDataAndRecordIndex(rliOpts, INSERT_OPERATION_OPT_VAL, SaveMode.Overwrite, + validate = false, numInserts = numSeedRecords) + + spark.sql(s"drop table if exists $sqlTable") + spark.sql(s"create table $sqlTable using hudi location '$basePath'") + + spark.sql("set hoodie.write.lock.provider = org.apache.hudi.client.transaction.lock.InProcessLockProvider") + spark.sql(s"set ${HoodieMetadataConfig.ENABLE.key} = true") + spark.sql(s"set ${HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key} = ${!isPartitionedRli}") + spark.sql(s"set ${HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key} = $isPartitionedRli") + spark.sql(s"set ${HoodieIndexConfig.INDEX_TYPE.key} = " + + (if (isPartitionedRli) "RECORD_LEVEL_INDEX" else "GLOBAL_RECORD_LEVEL_INDEX")) + + clearRliRegistry() + } + + /** + * The default path. Optimized writes make UPDATE a prepped write, so no index lookup happens and no + * counters are produced. Documented behaviour, asserted so it cannot change unnoticed. + */ + @Test + def testUpdateWithOptimizedWritesPerformsNoLookup(): Unit = { + seedTableAndRegisterForSql() + spark.sql(s"set ${SPARK_SQL_OPTIMIZED_WRITES.key} = true") + + spark.sql(s"update $sqlTable set rider = 'rider-optimized'") + + val counters = rliCountersFromLatestCommit() + report(s"Spark SQL UPDATE, optimized writes ON ($indexLabel) -- expected empty", counters) + assertTrue(counters.isEmpty, Review Comment: Resolved -- that test class is no longer in the PR. Spark SQL turned out not to report these counters at all, so rather than land a test that passes vacuously I removed the class and filed apache/hudi#19740 with its full source attached. ########## hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metrics/TestExecutorMetricsGenericity.java: ########## @@ -0,0 +1,170 @@ +/* + * 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.ExecutorMetricsContext; +import org.apache.hudi.common.metrics.LocalRegistry; +import org.apache.hudi.common.metrics.Registry; +import org.apache.hudi.config.HoodieWriteConfig; + +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * The claim this feature exists to support is that adding a class of executor metric costs a declaration + * plus two lines at the emission site. These tests hold that claim to a measurement rather than an + * argument, by collecting a group the shipping code has never heard of. + */ +public class TestExecutorMetricsGenericity { + + private static final String BASE_PATH = "file:///tmp/test_generic_metrics"; + + /** A class of metric added by a hypothetical future contributor: a name, a prefix, and no config. */ + private static final ExecutorMetricGroup STORAGE_CALLS = new ExecutorMetricGroup() { + @Override + public String registryName() { + return "HoodieStorageCalls"; + } + + @Override + public String commitMetadataPrefix() { + return "hoodie.storage.calls."; + } + + @Override + public String metricAction() { + return "storage"; + } + + @Override + public String metricQualifier() { + return "calls"; + } + + @Override + public boolean isEnabled(HoodieWriteConfig config) { + return true; + } + + @Override + public String scopedName(String basePath) { + return registryName() + ".test"; + } + }; + + private static HoodieWriteConfig config() { + return HoodieWriteConfig.newBuilder().withPath(BASE_PATH).forTable("generic_metrics_table").build(); + } + + private static Registry seed(HoodieWriteConfig cfg) { + Registry registry = new LocalRegistry(STORAGE_CALLS.scopedName(cfg.getBasePath())); + Registry.REGISTRY_MAP.put( Review Comment: Resolved -- `TestExecutorMetricsGenericity` is deleted along with the abstraction it existed to exercise. -- 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]
