This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 1112c37baf68 feat(metrics): report record index lookup counters to the
metrics reporter (#19575)
1112c37baf68 is described below
commit 1112c37baf68fbe48409046ad9b6ca003a846844
Author: Rahil C <[email protected]>
AuthorDate: Fri Aug 28 09:06:38 2026 -0700
feat(metrics): report record index lookup counters to the metrics reporter
(#19575)
There was no way to tell how much work a record index lookup did: how many
keys a commit looked up, what fraction were already in the table, or how
many
index file groups were read. Each task computes those numbers and returns
only the matches, and a miss produces no output row, so the driver cannot
recover them afterwards. HoodieMetadataMetrics has declared
lookup_record_index_key_count and lookup_record_index_key_hit_count for a
long time with nothing referencing them (see HUDI-9544). Part of #19063.
Every Spark write that tags records now reports, through the configured
metrics reporter and nothing else:
<table>.rli.lookup.lookup_record_index_key_count
<table>.rli.lookup.lookup_record_index_key_hit_count
<table>.rli.lookup.lookup_record_index_key_miss_count
<table>.rli.lookup.lookup_record_index_shards_read
<table>.rli.lookup.lookup_record_index_time
Counting is per record, not per distinct key, so hits + misses == key_count
holds exactly. Lookup time is summed across shards (they are read in
parallel), so it measures per-commit read effort; divide by shards_read for
a mean. Nothing is written to the timeline.
How it works:
- Counters are collected on executors into an AccumulatorV2-backed
DistributedRegistry. The driver resolves the registry
(RecordIndexLookupMetrics.resolveRegistry) and the lookup closure captures
it as a field, on both the global and partitioned record index paths;
nothing resolves a registry by name on the executor.
- Registries are owned by HoodieSparkEngineContext, keyed by a normalized
base path (authority + path, so Spark SQL's file:///data/t and the
DataSource's /data/t resolve to one entry while s3://a/t and s3://b/t stay
distinct). They are never published into Registry.REGISTRY_MAP; an entry
exists only between the lookup that created it and the commit that drains
it, so its presence is the record that this write looked something up.
- SparkRDDWriteClient.postCommit publishes and releases the registry once
the commit has landed, covering DataSource, Spark SQL DML and StreamSync
without a new hook on the base write client. A commit that never lands
publishes nothing, each commit reports only its own lookups (an abandoned
attempt's counters are discarded when the next write resolves the
registry), a commit that collected nothing zeroes the names it previously
published instead of re-reporting them, and publishing is wrapped in a
catch-all so a reporting problem cannot fail a completed write.
- INSERT drop-duplicates dedup now runs on the committing client's engine
context and config instead of a throwaway HoodieSparkEngineContext
(DataSourceUtils.resolveDuplicates is removed), so the lookup it performs
is attributed to the commit rather than stranded.
- DistributedRegistry ignores executor-side set/release instead of failing
the job, and failed task attempts are not merged, so a retry does not
inflate the counts. Duplicate successful attempts under speculation still
can; exactness is tracked in #19759.
Gated by the new hoodie.metrics.rli.lookup.enable (default false, not
inferred from hoodie.metrics.on, which it also requires). Off, nothing is
collected and no accumulator is registered. Spark only; counters aggregate
per table per JVM. hudi-io and BaseHoodieWriteClient are untouched;
hudi-client-common gains one config accessor.
Tests: unit coverage for the registry (including under task retry), the
config gate and RecordIndexLookupMetrics; functional coverage on the Spark
DataSource, Spark SQL, across a failed commit, and on the DeltaStreamer
path, all reading the counters off a capturing reporter. The accumulator
merge itself was certified with spark-submit on local-cluster[2,2,2560]
(two forked executor JVMs): 10000 / 5000 / 5000 across 10 shards on both of
two upsert rounds, the second reading 10000 rather than 20000.
---
.../org/apache/hudi/config/HoodieWriteConfig.java | 4 +
.../apache/hudi/client/SparkRDDWriteClient.java | 18 +-
.../client/common/HoodieSparkEngineContext.java | 32 +++
...titionedRecordIndexFileGroupLookupFunction.java | 19 ++
.../SparkMetadataTableGlobalRecordLevelIndex.java | 25 ++-
.../index/SparkMetadataTableRecordLevelIndex.java | 9 +-
.../apache/hudi/metrics/DistributedRegistry.java | 43 ++++
.../hudi/metrics/RecordIndexLookupMetrics.java | 218 +++++++++++++++++++++
.../hudi/metrics/TestDistributedRegistry.java | 102 ++++++++++
.../TestDistributedRegistryUnderTaskRetry.java | 148 ++++++++++++++
.../hudi/metrics/TestRecordIndexLookupMetrics.java | 215 ++++++++++++++++++++
.../common/config/metrics/HoodieMetricsConfig.java | 24 +++
.../config/metrics/TestHoodieMetricsConfig.java | 13 ++
.../main/java/org/apache/hudi/DataSourceUtils.java | 24 ---
.../org/apache/hudi/HoodieSparkSqlWriter.scala | 15 +-
.../hudi/PartitionedRecordLevelIndexSupport.scala | 1 +
.../java/org/apache/hudi/TestDataSourceUtils.java | 7 +-
.../hudi/testutils/CapturingMetricsReporter.java | 79 ++++++++
.../hudi/functional/RliLookupMetricsTestBase.scala | 136 +++++++++++++
.../TestRliLookupMetricsAcrossFailedCommit.scala | 125 ++++++++++++
.../TestRliLookupMetricsOnDataSource.scala | 154 +++++++++++++++
.../TestRliLookupMetricsOnSparkSql.scala | 138 +++++++++++++
.../TestRliMetricsOnStreamerPath.java | 128 ++++++++++++
.../testutils/CapturingMetricsReporter.java | 73 +++++++
24 files changed, 1708 insertions(+), 42 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
index c12acdbe2964..7bf46d9e9c33 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
@@ -2614,6 +2614,10 @@ public class HoodieWriteConfig extends HoodieConfig {
return metricsConfig.isLockingMetricsEnabled();
}
+ public boolean isRecordIndexLookupMetricsEnabled() {
+ return metricsConfig.isRecordIndexLookupMetricsEnabled();
+ }
+
public MetricsReporterType getMetricsReporterType() {
return metricsConfig.getMetricsReporterType();
}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
index ab331bdc7160..35c9abc76719 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
@@ -45,6 +45,7 @@ import org.apache.hudi.metadata.SparkMetadataWriterFactory;
import org.apache.hudi.metadata.StreamingMetadataWriteHandler;
import org.apache.hudi.metrics.DistributedRegistryUtil;
import org.apache.hudi.metrics.HoodieMetrics;
+import org.apache.hudi.metrics.RecordIndexLookupMetrics;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.hudi.table.HoodieSparkTable;
import org.apache.hudi.table.HoodieTable;
@@ -146,8 +147,8 @@ public class SparkRDDWriteClient<T> extends
// when streaming writes are enabled, writeStatuses is a mix of data
table write status and mdt write status
List<HoodieWriteStat> dataTableHoodieWriteStats =
slimWriteStatsList.stream().filter(entry ->
!entry.isMetadataTable()).map(SlimWriteStats::getWriteStat).collect(Collectors.toList());
List<HoodieWriteStat> partialMetadataTableWriteStats =
slimWriteStatsList.stream().filter(entry ->
entry.isMetadataTable).map(SlimWriteStats::getWriteStat).collect(Collectors.toList());
- return commitStats(instantTime, new
TableWriteStats(dataTableHoodieWriteStats, partialMetadataTableWriteStats),
extraMetadata, commitActionType, partitionToReplacedFileIds, extraPreCommitFunc,
- false, Option.of(table));
+ return commitStats(instantTime, new
TableWriteStats(dataTableHoodieWriteStats, partialMetadataTableWriteStats),
extraMetadata, commitActionType, partitionToReplacedFileIds,
+ extraPreCommitFunc, false, Option.of(table));
} else {
log.error("Exiting early due to errors with write operation ");
return false;
@@ -169,6 +170,19 @@ public class SparkRDDWriteClient<T> extends
}
}
+ /**
+ * The commit has landed, so the counters the executors collected for it can
be reported. A write that
+ * looked nothing up owns no registry and publishes nothing; a commit that
never lands never gets here.
+ */
+ @Override
+ protected void postCommit(HoodieTable table, HoodieCommitMetadata metadata,
String instantTime,
+ String commitActionType, Option<Map<String,
String>> extraMetadata) {
+ super.postCommit(table, metadata, instantTime, commitActionType,
extraMetadata);
+ if (config.isRecordIndexLookupMetricsEnabled()) {
+ RecordIndexLookupMetrics.publishAndRelease(context, config, metrics);
+ }
+ }
+
@Override
protected HoodieTable createTable(HoodieWriteConfig config) {
return createTableAndValidate(config, HoodieSparkTable::create);
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/HoodieSparkEngineContext.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/HoodieSparkEngineContext.java
index 091961bb4ca7..97317b61e490 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/HoodieSparkEngineContext.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/HoodieSparkEngineContext.java
@@ -288,6 +288,38 @@ public class HoodieSparkEngineContext extends
HoodieEngineContext {
});
}
+ /**
+ * Accumulator-backed registries owned by this context rather than by the
process, keyed by table base
+ * path. Unlike {@link #getMetricRegistry}, these are never published into
{@code Registry.REGISTRY_MAP}:
+ * the code that collects into them holds the registry by closure capture
and never resolves it by name,
+ * so a process-wide index buys nothing and costs a shared lifetime.
+ *
+ * <p>Ownership is what makes the counters attributable. The entry exists
only between the lookup that
+ * created it and the commit that drains it, so its presence is the record
that this write looked
+ * something up.
+ */
+ private final Map<String, DistributedRegistry> ownedRegistries = new
ConcurrentHashMap<>();
+
+ /**
+ * The registry a write collects into, created and registered with this
context's {@code SparkContext}
+ * on first use. Callers drain it with {@link #removeOwnedRegistry}.
+ */
+ public DistributedRegistry getOrCreateOwnedRegistry(String key, String
registryName) {
+ return ownedRegistries.computeIfAbsent(key, k -> {
+ DistributedRegistry registry = new DistributedRegistry(registryName);
+ registry.register(javaSparkContext);
+ return registry;
+ });
+ }
+
+ /**
+ * Removes and returns the registry for a key, or empty when this context
never created one. Empty is
+ * how a write that performed no lookup is distinguished from one that did.
+ */
+ public Option<DistributedRegistry> removeOwnedRegistry(String key) {
+ return Option.ofNullable(ownedRegistries.remove(key));
+ }
+
/**
* Register the distributed registries on Spark executors.
* This is called within Spark operations to make the registries available
on executors.
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/PartitionedRecordIndexFileGroupLookupFunction.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/PartitionedRecordIndexFileGroupLookupFunction.java
index c32dd7706e95..d9096655f962 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/PartitionedRecordIndexFileGroupLookupFunction.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/PartitionedRecordIndexFileGroupLookupFunction.java
@@ -21,10 +21,13 @@ package org.apache.hudi.index;
import org.apache.hudi.common.data.HoodieListData;
import org.apache.hudi.common.data.HoodiePairData;
+import org.apache.hudi.common.metrics.Registry;
import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
+import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.metrics.RecordIndexLookupMetrics;
import org.apache.spark.api.java.function.PairFlatMapFunction;
@@ -44,9 +47,18 @@ public class PartitionedRecordIndexFileGroupLookupFunction
implements PairFlatMapFunction<Iterator<Pair<String, String>>, String,
HoodieRecordGlobalLocation> {
private final HoodieTableMetadata metadataTable;
+ /** Empty when no counters should be collected; see
RecordIndexLookupMetrics#resolveRegistry. */
+ private final Option<Registry> lookupMetrics;
+ /** Uninstrumented, for the query-side read path. */
public PartitionedRecordIndexFileGroupLookupFunction(HoodieTableMetadata
metadataTable) {
+ this(metadataTable, Option.empty());
+ }
+
+ public PartitionedRecordIndexFileGroupLookupFunction(HoodieTableMetadata
metadataTable,
+ Option<Registry>
lookupMetrics) {
this.metadataTable = metadataTable;
+ this.lookupMetrics = lookupMetrics;
}
@Override
@@ -65,11 +77,18 @@ public class PartitionedRecordIndexFileGroupLookupFunction
return Collections.emptyIterator();
}
+ // Started only when collecting: an unused timer is an allocation per
shard on the disabled path.
+ HoodieTimer shardTimer = lookupMetrics.isPresent() ? HoodieTimer.start() :
null;
HoodiePairData<String, HoodieRecordGlobalLocation> recordIndexData =
metadataTable.readRecordIndexLocationsWithKeys(HoodieListData.eager(keysToLookup),
Option.of(partitionName));
try {
Map<String, HoodieRecordGlobalLocation> recordIndexInfo =
recordIndexData.collectAsList().stream()
.collect(HashMap::new, (map, pair) -> map.put(pair.getKey(),
pair.getValue()), HashMap::putAll);
+ // recordIndexInfo is keyed by record key, so its key set is the found
set with no extra allocation.
+ if (lookupMetrics.isPresent()) {
+ RecordIndexLookupMetrics.recordShardLookup(lookupMetrics.get(),
keysToLookup,
+ recordIndexInfo.keySet(), shardTimer.endTimer());
+ }
return recordIndexInfo.entrySet().stream()
.map(e -> new Tuple2<>(e.getKey(), e.getValue())).iterator();
} finally {
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/SparkMetadataTableGlobalRecordLevelIndex.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/SparkMetadataTableGlobalRecordLevelIndex.java
index 31150948b54f..88499c69fb4b 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/SparkMetadataTableGlobalRecordLevelIndex.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/SparkMetadataTableGlobalRecordLevelIndex.java
@@ -24,10 +24,13 @@ import org.apache.hudi.common.data.HoodieListData;
import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.function.SerializableBiFunction;
+import org.apache.hudi.common.metrics.Registry;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
import org.apache.hudi.common.util.Either;
import org.apache.hudi.common.util.HoodieDataUtils;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieIndexConfig;
@@ -38,6 +41,7 @@ import org.apache.hudi.exception.HoodieIndexException;
import org.apache.hudi.exception.TableNotFoundException;
import org.apache.hudi.metadata.HoodieIndexVersion;
import org.apache.hudi.metadata.MetadataPartitionType;
+import org.apache.hudi.metrics.RecordIndexLookupMetrics;
import org.apache.hudi.table.HoodieTable;
import lombok.extern.slf4j.Slf4j;
@@ -49,6 +53,7 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
import scala.Tuple2;
@@ -128,9 +133,10 @@ public class SparkMetadataTableGlobalRecordLevelIndex
extends HoodieIndex<Object
// keyToLocationPairRDD and records RDD.
ValidationUtils.checkState(partitionedKeyRDD.getNumPartitions() <=
numFileGroups);
- // Lookup the keys in the record index
-
- return HoodieJavaPairRDD.of(partitionedKeyRDD.mapPartitionsToPair(new
RecordIndexFileGroupLookupFunction(hoodieTable)));
+ // Resolved on the driver so the closure carries it to executors.
+ Option<Registry> lookupMetrics =
RecordIndexLookupMetrics.resolveRegistry(context, hoodieTable.getConfig());
+ return HoodieJavaPairRDD.of(partitionedKeyRDD.mapPartitionsToPair(
+ new RecordIndexFileGroupLookupFunction(hoodieTable, lookupMetrics)));
}
protected Either<Integer, Map<String, Integer>>
fetchFileGroupSize(HoodieTable hoodieTable) {
@@ -181,9 +187,12 @@ public class SparkMetadataTableGlobalRecordLevelIndex
extends HoodieIndex<Object
*/
private static class RecordIndexFileGroupLookupFunction implements
PairFlatMapFunction<Iterator<String>, String, HoodieRecordGlobalLocation> {
private final HoodieTable hoodieTable;
+ /** Empty when no counters should be collected; see
RecordIndexLookupMetrics#resolveRegistry. */
+ private final Option<Registry> lookupMetrics;
- public RecordIndexFileGroupLookupFunction(HoodieTable hoodieTable) {
+ public RecordIndexFileGroupLookupFunction(HoodieTable hoodieTable,
Option<Registry> lookupMetrics) {
this.hoodieTable = hoodieTable;
+ this.lookupMetrics = lookupMetrics;
}
@Override
@@ -191,11 +200,19 @@ public class SparkMetadataTableGlobalRecordLevelIndex
extends HoodieIndex<Object
List<String> keysToLookup = new ArrayList<>();
recordKeyIterator.forEachRemaining(keysToLookup::add);
+ // Started only when collecting: an unused timer is an allocation per
shard on the disabled path.
+ HoodieTimer shardTimer = lookupMetrics.isPresent() ? HoodieTimer.start()
: null;
// recordIndexInfo object only contains records that are present in
record_index.
HoodiePairData<String, HoodieRecordGlobalLocation> recordIndexData =
hoodieTable.getTableMetadata().readRecordIndexLocationsWithKeys(HoodieListData.eager(keysToLookup));
try {
List<Pair<String, HoodieRecordGlobalLocation>> recordIndexInfo =
HoodieDataUtils.dedupeAndCollectAsList(recordIndexData);
+ // Guarded rather than checked inside the helper: the found set is
O(hits) and Java evaluates it
+ // as an argument first, so an unguarded call would cost every shard
that on the disabled path.
+ if (lookupMetrics.isPresent()) {
+ RecordIndexLookupMetrics.recordShardLookup(lookupMetrics.get(),
keysToLookup,
+
recordIndexInfo.stream().map(Pair::getKey).collect(Collectors.toSet()),
shardTimer.endTimer());
+ }
return recordIndexInfo.stream()
.map(e -> new Tuple2<>(e.getKey(), e.getValue())).iterator();
} finally {
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/SparkMetadataTableRecordLevelIndex.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/SparkMetadataTableRecordLevelIndex.java
index 6a12dc3c67a5..d98c724ea96f 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/SparkMetadataTableRecordLevelIndex.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/SparkMetadataTableRecordLevelIndex.java
@@ -22,10 +22,12 @@ package org.apache.hudi.index;
import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.metrics.Registry;
import org.apache.hudi.common.model.FileSlice;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
import org.apache.hudi.common.util.Either;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
@@ -34,6 +36,7 @@ import org.apache.hudi.data.HoodieJavaRDD;
import org.apache.hudi.metadata.BucketizedMetadataTableFileGroupIndexParser;
import org.apache.hudi.metadata.HoodieTableMetadataUtil;
import org.apache.hudi.metadata.MetadataPartitionType;
+import org.apache.hudi.metrics.RecordIndexLookupMetrics;
import org.apache.hudi.table.HoodieTable;
import org.apache.spark.api.java.JavaRDD;
@@ -77,8 +80,10 @@ public class SparkMetadataTableRecordLevelIndex extends
SparkMetadataTableGlobal
.partitionBy(new PartitionIdPassthrough(numFileGroups))
.map(t -> t._2);
ValidationUtils.checkState(partitionedKeyRDD.getNumPartitions() <=
numFileGroups);
- // Lookup the keys in the record index
- return HoodieJavaPairRDD.of(partitionedKeyRDD.mapPartitionsToPair(new
PartitionedRecordIndexFileGroupLookupFunction(hoodieTable.getTableMetadata())));
+ // Lookup the keys in the record index. Resolved on the driver so the
closure carries it to executors.
+ Option<Registry> lookupMetrics =
RecordIndexLookupMetrics.resolveRegistry(context, hoodieTable.getConfig());
+ return HoodieJavaPairRDD.of(partitionedKeyRDD.mapPartitionsToPair(
+ new
PartitionedRecordIndexFileGroupLookupFunction(hoodieTable.getTableMetadata(),
lookupMetrics)));
}
@Override
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metrics/DistributedRegistry.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metrics/DistributedRegistry.java
index 4da400d8869a..c8b9fdde6458 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metrics/DistributedRegistry.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metrics/DistributedRegistry.java
@@ -20,8 +20,11 @@ package org.apache.hudi.metrics;
import org.apache.hudi.common.metrics.Registry;
+import org.apache.spark.TaskContext;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.util.AccumulatorV2;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.Serializable;
import java.util.HashMap;
@@ -33,8 +36,13 @@ import java.util.concurrent.ConcurrentHashMap;
*/
public class DistributedRegistry extends AccumulatorV2<Map<String, Long>,
Map<String, Long>>
implements Registry, Serializable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(DistributedRegistry.class);
+
private final String name;
ConcurrentHashMap<String, Long> counters = new ConcurrentHashMap<>();
+ /** Driver-only, to detect a SparkContext restart in the same JVM (shells,
notebooks, Spark Connect). */
+ private transient String registeredAppId;
public DistributedRegistry(String name) {
this.name = name;
@@ -48,9 +56,17 @@ public class DistributedRegistry extends
AccumulatorV2<Map<String, Long>, Map<St
public void register(JavaSparkContext jsc) {
if (!isRegistered()) {
jsc.sc().register(this);
+ // Only when this call actually registers: stamping unconditionally
would re-brand an accumulator
+ // bound to a dead context and mask the staleness this field exists to
detect.
+ this.registeredAppId = jsc.sc().applicationId();
}
}
+ /** False when bound to a different (typically stopped) context, meaning it
must be recreated. */
+ public boolean isRegisteredWith(JavaSparkContext jsc) {
+ return isRegistered() && jsc.sc().applicationId().equals(registeredAppId);
+ }
+
@Override
public void clear() {
counters.clear();
@@ -68,9 +84,36 @@ public class DistributedRegistry extends
AccumulatorV2<Map<String, Long>, Map<St
@Override
public void set(String name, long value) {
+ // Last-writer-wins is neither commutative nor associative, and the driver
merges executor copies in
+ // an unspecified order. Driver only; executors use increment()/add().
+ if (TaskContext.get() != null) {
+ // Warn rather than throw: this runs inside a task, and a metrics
problem must not fail a write.
+ LOG.warn("DistributedRegistry.set() called from a Spark executor and
ignored: it is non-commutative "
+ + "under accumulator merges and would produce non-deterministic
values. Use increment()/add().");
+ return;
+ }
counters.merge(name, value, (oldValue, newValue) -> newValue);
}
+ /**
+ * Subtracts rather than clearing, so a concurrent merge either lands before
the subtraction and is
+ * released with it, or after and survives. Clamped at zero because {@code
Metrics.shutdown()} can empty
+ * the registry underneath a release; counters reaching zero are removed so
a table that performed no
+ * lookup is distinguishable from one that missed everything.
+ */
+ public void release(Map<String, Long> counts) {
+ // Driver-only for the same reason as set(): clamping and eviction are
order-dependent under merges.
+ if (TaskContext.get() != null) {
+ LOG.warn("DistributedRegistry.release() called from a Spark executor and
ignored: clamping and eviction "
+ + "are order-dependent under accumulator merges. Release at the
commit boundary on the driver.");
+ return;
+ }
+ counts.forEach((name, released) -> counters.compute(name, (key, current)
-> {
+ long remaining = (current == null ? 0L : current) - released;
+ return remaining > 0L ? remaining : null;
+ }));
+ }
+
/**
* Get all Counter type metrics.
*/
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metrics/RecordIndexLookupMetrics.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metrics/RecordIndexLookupMetrics.java
new file mode 100644
index 000000000000..e901c51a2554
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metrics/RecordIndexLookupMetrics.java
@@ -0,0 +1,218 @@
+/*
+ * 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.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.storage.StoragePath;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The record index lookup counters, end to end: what they are called, where
they are collected, and how
+ * they reach the reporter once a commit lands.
+ *
+ * <p>The numbers are only knowable on executors, because a lookup returns its
hits and a miss produces
+ * no output row at all. So the driver resolves an accumulator-backed {@link
DistributedRegistry}, the
+ * lookup closure captures it, tasks add into their own copy, and Spark merges
those copies home.
+ */
+public class RecordIndexLookupMetrics {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(RecordIndexLookupMetrics.class);
+
+ /** The registry name, as reported. Keying is done by {@link #registryKey}.
*/
+ public static final String REGISTRY_NAME = "HoodieRecordIndexLookup";
+
+ /** Reporter naming, as passed to {@code HoodieMetrics.getMetricsName}. */
+ public static final String METRIC_ACTION = "rli";
+ public static final String METRIC_QUALIFIER = "lookup";
+
+ /** Counts records, not distinct keys: a batch repeating a key contributes
once per record, which is
+ * what keeps {@code hits + misses == key_count} exact. */
+ 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 rather than
averaged because shards are read
+ * in parallel: the value is per-commit read effort, and dividing by {@link
#SHARDS_READ} gives a mean.
+ *
+ * <p>Distinct from {@code index.lookup.duration} published by {@code
HoodieMetrics.updateIndexMetrics},
+ * which is driver wall-clock for the whole {@code tagLocation} including
scheduling. Comparing the two
+ * shows how much of a lookup was actually spent reading the index.
+ */
+ public static final String LOOKUP_TIME = "lookup_record_index_time";
+
+ private RecordIndexLookupMetrics() {
+ }
+
+ /**
+ * The registry a lookup task collects into, or null when nothing should be
collected. Captured in the
+ * lookup closure and passed to {@link #recordShardLookup}, so delivery is
by closure capture rather
+ * than by name.
+ *
+ * <p>Requires the reporter to be on as well: with {@code hoodie.metrics.on}
off there is nowhere to
+ * publish, and collecting would register an accumulator and scan every
shard for nothing.
+ */
+ public static Option<Registry> resolveRegistry(HoodieEngineContext context,
HoodieWriteConfig config) {
+ if (!config.isMetricsOn() || !config.isRecordIndexLookupMetricsEnabled()
+ || !(context instanceof HoodieSparkEngineContext)) {
+ return Option.empty();
+ }
+ DistributedRegistry registry = ((HoodieSparkEngineContext) context)
+ .getOrCreateOwnedRegistry(registryKey(config.getBasePath()),
REGISTRY_NAME);
+ // Anything held predates this write: the drain removes the entry as it
publishes, so a surviving
+ // one belongs to an attempt that never committed.
+ registry.clear();
+ return Option.of(registry);
+ }
+
+ /**
+ * Records one shard's lookup outcome. Counts records rather than distinct
keys, so
+ * {@code hits + misses == records_looked_up} holds when a batch repeats a
key. Membership is tested
+ * against the found set, bounded by the hit count, not the asked-about set,
bounded by shard size.
+ *
+ * <p>Callers must skip the call entirely when the registry is null.
Building the found set is
+ * O(hits), and an argument is evaluated before this method can take a null
fast path, so guarding
+ * here would not keep that cost off the disabled path.
+ *
+ * @param registry where to collect, never null
+ * @param keysLookedUp every record key routed to this shard
+ * @param foundKeys the subset present in the index
+ * @param elapsedMs wall-clock spent reading this shard
+ */
+ public static void recordShardLookup(Registry registry, Collection<String>
keysLookedUp,
+ Collection<String> foundKeys, long
elapsedMs) {
+ if (keysLookedUp.isEmpty()) {
+ return;
+ }
+ Set<String> found = foundKeys instanceof Set ? (Set<String>) foundKeys :
new HashSet<>(foundKeys);
+ long records = keysLookedUp.size();
+ long hits = found.isEmpty() ? 0L :
keysLookedUp.stream().filter(found::contains).count();
+ registry.add(KEY_COUNT, records);
+ registry.add(KEY_HIT_COUNT, hits);
+ registry.add(KEY_MISS_COUNT, records - hits);
+ registry.increment(SHARDS_READ);
+ registry.add(LOOKUP_TIME, elapsedMs);
+ }
+
+ /**
+ * Reports the counters and releases them, so the next commit reports only
its own work. Called once
+ * the commit has landed, so a commit that never lands publishes nothing.
+ *
+ * <p>Release subtracts what was reported rather than clearing, so a
straggler task whose update lands
+ * mid-publish carries into the next commit instead of being dropped. An
all-zero registry is skipped.
+ */
+ public static void publishAndRelease(HoodieEngineContext context,
HoodieWriteConfig config,
+ HoodieMetrics hoodieMetrics) {
+ try {
+ publish(context, config, hoodieMetrics);
+ } catch (Exception e) {
+ // This runs after the commit has landed. Reporting is not worth failing
a completed write over.
+ LOG.warn("Failed to publish record index lookup metrics; the commit is
unaffected.", e);
+ }
+ }
+
+ private static void publish(HoodieEngineContext context, HoodieWriteConfig
config,
+ HoodieMetrics hoodieMetrics) {
+ if (!config.isRecordIndexLookupMetricsEnabled()) {
+ // Callers gate before reaching here, so this is a wiring mistake rather
than normal flow.
+ LOG.warn("Record index lookup metrics drain reached with the feature
disabled; nothing published.");
+ return;
+ }
+ if (!(context instanceof HoodieSparkEngineContext)) {
+ return;
+ }
+ // Removing is what makes attribution work. An entry exists only because a
lookup on this write
+ // created it, so its absence means this write looked nothing up and has
nothing of its own to
+ // report -- which is the case an operation type cannot tell apart, since
an insert that drops
+ // duplicates tags through SparkRDDReadClient without requiring tagging.
+ Option<DistributedRegistry> taken =
+ ((HoodieSparkEngineContext)
context).removeOwnedRegistry(registryKey(config.getBasePath()));
+ if (!taken.isPresent()) {
+ zeroPreviouslyReported(hoodieMetrics);
+ return;
+ }
+ DistributedRegistry registry = taken.get();
+ Map<String, Long> counts = new HashMap<>(registry.getAllCounts(false));
+ if (counts.values().stream().allMatch(value -> value == 0L)) {
+ // Gauges hold their last value until overwritten, so leaving them alone
would have a reporter
+ // re-emit the previous commit's numbers for this one. Zero them instead.
+ zeroPreviouslyReported(hoodieMetrics);
+ return;
+ }
+ publishToReporter(counts, hoodieMetrics);
+ registry.release(counts);
+ }
+
+ /**
+ * Resets the gauges to zero, for a commit that collected nothing. Only
names already published are
+ * touched, so a table that has never emitted stays absent rather than
reporting a row of zeros.
+ */
+ private static void zeroPreviouslyReported(HoodieMetrics hoodieMetrics) {
+ if (hoodieMetrics == null || hoodieMetrics.getMetrics() == null) {
+ return;
+ }
+ String prefix = hoodieMetrics.getMetricsName(METRIC_ACTION,
METRIC_QUALIFIER);
+ if (prefix == null) {
+ return;
+ }
+ Map<String, Long> zeroed = new HashMap<>();
+ hoodieMetrics.getMetrics().getRegistry().getGauges().keySet().stream()
+ .filter(name -> name.startsWith(prefix + "."))
+ .forEach(name -> zeroed.put(name.substring(prefix.length() + 1), 0L));
+ if (!zeroed.isEmpty()) {
+ publishToReporter(zeroed, hoodieMetrics);
+ }
+ }
+
+ /** Gauges, so each commit overwrites the previous value rather than
accumulating. */
+ private static void publishToReporter(Map<String, Long> counts,
HoodieMetrics hoodieMetrics) {
+ if (hoodieMetrics == null || hoodieMetrics.getMetrics() == null) {
+ return;
+ }
+ String prefix = hoodieMetrics.getMetricsName(METRIC_ACTION,
METRIC_QUALIFIER);
+ hoodieMetrics.getMetrics().registerGauges(counts,
Option.ofNullable(prefix));
+ }
+
+ /**
+ * Key for this table's registry inside the owning context. The base path is
normalized rather than used
+ * raw because one table is spelled more than one way: Spark SQL builds its
config from the catalog
+ * location ({@code file:///data/t}) while the DataSource passes the bare
path ({@code /data/t}), and
+ * the two must resolve to one entry. The authority is kept so {@code
s3://a/t} and {@code s3://b/t}
+ * stay distinct.
+ *
+ * <p>No digest: this key never leaves the context and never becomes part of
a metric name.
+ */
+ static String registryKey(String basePath) {
+ StoragePath path = new StoragePath(basePath);
+ String authority = path.toUri().getAuthority();
+ return (authority == null ? "" : authority) +
path.getPathWithoutSchemeAndAuthority();
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestDistributedRegistry.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestDistributedRegistry.java
index c64e11a27059..6058eeef3522 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestDistributedRegistry.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestDistributedRegistry.java
@@ -29,6 +29,7 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -184,6 +185,107 @@ public class TestDistributedRegistry {
Assertions.assertEquals(finalExpectedSum, metricCounts.get(METRIC_1));
}
+ @Test
+ public void testSetOnExecutorIsIgnoredWithoutFailingTheJob() {
+ // 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: set() is invoked from an executor. It is non-commutative under
accumulator merges, so the
+ // value must not be recorded -- but the guard runs inside a task, where
failing would take the write
+ // down with it, so it is ignored rather than thrown.
+ engineContext.map(data, value -> {
+ registry.set(METRIC_1, value);
+ return null;
+ }, 1);
+
+ // Then: the job succeeded and nothing was recorded.
+ Assertions.assertFalse(registry.getAllCounts().containsKey(METRIC_1),
+ "set() from an executor must not record a value: " +
registry.getAllCounts());
+ }
+
+ @Test
+ public void testReleaseOnExecutorIsIgnoredWithoutFailingTheJob() {
+ // Given: a registry holding a known count
+ String registryName = REGISTRY_NAME + "_testReleaseOnExecutor";
+ DistributedRegistry registry = (DistributedRegistry)
engineContext.getMetricRegistry("", registryName);
+ registry.add(METRIC_1, 5L);
+
+ List<Integer> data = new ArrayList<>();
+ data.add(1);
+
+ // When: release() is invoked from an executor. Clamping and eviction are
order-dependent under
+ // accumulator merges, and this runs after the commit has landed, so it is
ignored rather than thrown.
+ engineContext.map(data, value -> {
+ registry.release(Collections.singletonMap(METRIC_1, (long) value));
+ return null;
+ }, 1);
+
+ // Then: the job succeeded and the count was left alone.
+ Assertions.assertEquals(5L, registry.getAllCounts().get(METRIC_1),
+ "release() from an executor must not mutate the counters");
+ }
+
+ @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 testReleaseEvictsCountersThatReachZero() {
+ // Given: a registry drained of exactly what it holds, as the
commit-boundary drain does
+ DistributedRegistry registry = new DistributedRegistry(REGISTRY_NAME +
"_testRelease");
+ registry.add(METRIC_1, 10);
+ registry.add(METRIC_2, 20);
+
+ // When: the full contents are released
+ registry.release(registry.getAllCounts(false));
+
+ // Then: the counters are gone, not left sitting at zero. A zero-valued
entry survives
+ // ConcurrentHashMap.merge() and would be republished by every later drain.
+ Assertions.assertTrue(registry.getAllCounts().isEmpty(),
+ "released counters must be evicted, found " + registry.getAllCounts());
+ Assertions.assertTrue(registry.isZero());
+ }
+
+ @Test
+ public void testReleaseKeepsWhatArrivedAfterTheDrain() {
+ // Given: a snapshot taken, and a straggler update folded in before the
release lands
+ DistributedRegistry registry = new DistributedRegistry(REGISTRY_NAME +
"_testReleaseStraggler");
+ registry.add(METRIC_1, 10);
+ Map<String, Long> drained = registry.getAllCounts(false);
+ registry.add(METRIC_1, 4);
+
+ // When: the drained counts are released
+ registry.release(drained);
+
+ // Then: only the drained amount is subtracted - the straggler belongs to
the next drain
+ Assertions.assertEquals(4L, registry.getAllCounts().get(METRIC_1));
+ }
+
+ @Test
+ public void testReleaseClampsAtZero() {
+ // Given: a registry emptied between the drain and the release.
Metrics.shutdown() scrapes every
+ // registry in the process with flush=true, so an unrelated table's write
finishing does exactly this.
+ DistributedRegistry registry = new DistributedRegistry(REGISTRY_NAME +
"_testReleaseClamp");
+ registry.add(METRIC_1, 10);
+ Map<String, Long> drained = registry.getAllCounts(false);
+ registry.clear();
+
+ // When: the release lands on a registry that no longer holds those counts
+ registry.release(drained);
+
+ // Then: nothing negative is left behind - a negative counter would be
stamped into commit metadata
+ Assertions.assertTrue(registry.getAllCounts().isEmpty(),
+ "an unbounded subtraction would have left " + METRIC_1 + " negative,
got " + registry.getAllCounts());
+ }
+
@Test
public void testClear() {
// Given: distributed registry with some metrics
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestDistributedRegistryUnderTaskRetry.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestDistributedRegistryUnderTaskRetry.java
new file mode 100644
index 000000000000..895c07bdeaae
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestDistributedRegistryUnderTaskRetry.java
@@ -0,0 +1,148 @@
+/*
+ * 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.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.common.metrics.Registry;
+
+import org.apache.spark.SparkConf;
+import org.apache.spark.TaskContext;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** What task retries and stage recomputation do to counters collected in a
{@link DistributedRegistry}. */
+public class TestDistributedRegistryUnderTaskRetry {
+
+ private static final String COUNTER = "keys";
+
+ private JavaSparkContext jsc;
+
+ @AfterEach
+ public void tearDown() {
+ if (jsc != null) {
+ jsc.stop();
+ jsc = null;
+ }
+ }
+
+ /** {@code local[n, maxFailures]} -- the second number is what makes Spark
re-attempt a failed task. */
+ private JavaSparkContext retryingContext(String appName) {
+ SparkConf conf = new SparkConf()
+ .setAppName(appName)
+ .setMaster("local[2, 4]")
+ .set("spark.ui.enabled", "false")
+ .set("spark.sql.shuffle.partitions", "4");
+ return new JavaSparkContext(conf);
+ }
+
+ private static List<Integer> partitionedInput(int numPartitions, int
perPartition) {
+ List<Integer> data = new ArrayList<>();
+ for (int i = 0; i < numPartitions * perPartition; i++) {
+ data.add(i);
+ }
+ return data;
+ }
+
+ /**
+ * One task fails its first attempt and succeeds on the retry. The count
must reflect one pass over
+ * the data, not two -- Spark discards the failed attempt's accumulator
updates.
+ */
+ @Test
+ public void testFailedAttemptDoesNotInflateCounts() {
+ jsc = retryingContext("rli-retry-failed-attempt");
+ HoodieSparkEngineContext context = new HoodieSparkEngineContext(jsc);
+ Registry registry = context.getOrCreateOwnedRegistry("retryRegistry",
"retryRegistry");
+
+ int numPartitions = 4;
+ int perPartition = 25;
+
+ long collected = jsc.parallelize(partitionedInput(numPartitions,
perPartition), numPartitions)
+ .mapPartitions(it -> {
+ List<Integer> batch = new ArrayList<>();
+ it.forEachRemaining(batch::add);
+ registry.add(COUNTER, batch.size());
+ // Partition 0 blows up the first time it is tried. maxFailures=4
lets the retry through.
+ if (TaskContext.getPartitionId() == 0 &&
TaskContext.get().attemptNumber() == 0) {
+ throw new RuntimeException("induced failure on first attempt of
partition 0");
+ }
+ return Collections.singletonList((long) batch.size()).iterator();
+ })
+ .reduce(Long::sum);
+
+ long counted = registry.getAllCounts(false).get(COUNTER);
+ long expected = (long) numPartitions * perPartition;
+
+ System.out.println("\n===== failed attempt =====");
+ System.out.println(" rows actually processed " + collected);
+ System.out.println(" registry counted " + counted);
+ System.out.println(" expected " + expected);
+ System.out.println("==========================\n");
+
+ assertEquals(expected, collected, "sanity: the retry must have succeeded
and produced every row");
+ assertEquals(expected, counted,
+ "a failed attempt must contribute nothing: Spark ships accumulator
updates home only from "
+ + "attempts that succeed, which is what makes plain task retries
safe for these counters");
+ }
+
+ /**
+ * The other half, asserted so the caveat is grounded rather than asserted
in prose: evaluating the same uncached transformation twice counts twice.
+ */
+ @Test
+ public void testRepeatedEvaluationDoubleCounts() {
+ jsc = retryingContext("rli-retry-recompute");
+ HoodieSparkEngineContext context = new HoodieSparkEngineContext(jsc);
+ Registry registry = context.getOrCreateOwnedRegistry("recomputeRegistry",
"recomputeRegistry");
+
+ int numPartitions = 4;
+ int perPartition = 25;
+ long onePass = (long) numPartitions * perPartition;
+
+ org.apache.spark.api.java.JavaRDD<Long> counted =
+ jsc.parallelize(partitionedInput(numPartitions, perPartition),
numPartitions)
+ .mapPartitions(it -> {
+ List<Integer> batch = new ArrayList<>();
+ it.forEachRemaining(batch::add);
+ registry.add(COUNTER, batch.size());
+ return Collections.singletonList((long) batch.size()).iterator();
+ });
+
+ // Two actions over an uncached RDD: the transformation runs twice.
+ counted.reduce(Long::sum);
+ counted.reduce(Long::sum);
+
+ long total = registry.getAllCounts(false).get(COUNTER);
+
+ System.out.println("\n===== repeated evaluation =====");
+ System.out.println(" one pass would be " + onePass);
+ System.out.println(" registry counted " + total);
+ System.out.println("===============================\n");
+
+ assertEquals(onePass * 2, total,
+ "counters incremented inside a transformation are at-least-once: a
second evaluation of the "
+ + "same uncached RDD counts again. Documented behaviour, pinned
here so a change in either "
+ + "direction is visible");
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestRecordIndexLookupMetrics.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestRecordIndexLookupMetrics.java
new file mode 100644
index 000000000000..35214f4aef48
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestRecordIndexLookupMetrics.java
@@ -0,0 +1,215 @@
+/*
+ * 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.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.common.config.metrics.HoodieMetricsConfig;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.testutils.HoodieClientTestUtils;
+
+import org.apache.spark.api.java.JavaSparkContext;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.apache.hudi.metrics.RecordIndexLookupMetrics.KEY_COUNT;
+import static org.apache.hudi.metrics.RecordIndexLookupMetrics.KEY_HIT_COUNT;
+import static org.apache.hudi.metrics.RecordIndexLookupMetrics.KEY_MISS_COUNT;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Drain semantics for the record index lookup counters, at the level where
they are decided. */
+public class TestRecordIndexLookupMetrics {
+
+ private static final String TABLE = "drain_test_table";
+ private static final String BASE_PATH = "file:/tmp/drain_test_table";
+ private static final String OTHER_BASE_PATH = "file:/tmp/somewhere_else";
+
+ private static final String LOOKED_UP_KEY = KEY_COUNT;
+
+ private HoodieMetrics hoodieMetrics;
+ private static JavaSparkContext jsc;
+ private static HoodieSparkEngineContext context;
+
+ @BeforeAll
+ static void startContext() {
+ jsc = new
JavaSparkContext(HoodieClientTestUtils.getSparkConfForTest("drain-test"));
+ context = new HoodieSparkEngineContext(jsc);
+ }
+
+ @AfterAll
+ static void stopContext() {
+ if (jsc != null) {
+ jsc.stop();
+ jsc = null;
+ }
+ }
+
+ @AfterEach
+ void clearOwnedRegistries() {
+ // Registries live on the context now, so draining what a test left is all
the cleanup there is.
+
context.removeOwnedRegistry(RecordIndexLookupMetrics.registryKey(BASE_PATH));
+
context.removeOwnedRegistry(RecordIndexLookupMetrics.registryKey(OTHER_BASE_PATH));
+ if (hoodieMetrics != null && hoodieMetrics.getMetrics() != null) {
+ hoodieMetrics.getMetrics().shutdown();
+ hoodieMetrics = null;
+ }
+ }
+
+ private static HoodieWriteConfig config(String basePath, boolean gateOn) {
+ return HoodieWriteConfig.newBuilder()
+ .withPath(basePath)
+ .forTable(TABLE)
+ .withMetricsConfig(HoodieMetricsConfig.newBuilder()
+ .on(true)
+ .withReporterType(MetricsReporterType.INMEMORY.name())
+ .withRecordIndexLookupMetrics(gateOn)
+ .build())
+ .build();
+ }
+
+ /** The reporter this drain publishes into, for the table under test. */
+ private HoodieMetrics metricsFor(HoodieWriteConfig config) {
+ hoodieMetrics = new HoodieMetrics(config,
HoodieTestUtils.getDefaultStorage());
+ return hoodieMetrics;
+ }
+
+ /** Creates the entry the way a lookup does, then fills it as the executors
would. */
+ private static Registry seedRegistry(HoodieWriteConfig config, long
lookedUp, long hits, long misses) {
+ Registry registry = context.getOrCreateOwnedRegistry(
+ RecordIndexLookupMetrics.registryKey(config.getBasePath()),
RecordIndexLookupMetrics.REGISTRY_NAME);
+ registry.add(LOOKED_UP_KEY, lookedUp);
+ registry.add(KEY_HIT_COUNT, hits);
+ registry.add(KEY_MISS_COUNT, misses);
+ return registry;
+ }
+
+ /** Gauge name the reporter publishes a counter under. */
+ private String gaugeName(HoodieMetrics metrics, String metric) {
+ return metrics.getMetricsName(
+ RecordIndexLookupMetrics.METRIC_ACTION,
+ RecordIndexLookupMetrics.METRIC_QUALIFIER)
+ + "." + metric;
+ }
+
+ private Long gauge(HoodieMetrics metrics, String metric) {
+ Map<String, com.codahale.metrics.Gauge> gauges =
metrics.getMetrics().getRegistry().getGauges();
+ com.codahale.metrics.Gauge<?> g = gauges.get(gaugeName(metrics, metric));
+ return g == null ? null : (Long) g.getValue();
+ }
+
+ @Test
+ void publishReportsWhatTheRegistryHolds() {
+ HoodieWriteConfig config = config(BASE_PATH, true);
+ seedRegistry(config, 10L, 7L, 3L);
+ HoodieMetrics metrics = metricsFor(config);
+
+ RecordIndexLookupMetrics.publishAndRelease(context, config, metrics);
+
+ assertEquals(10L, gauge(metrics, KEY_COUNT));
+ assertEquals(7L, gauge(metrics, KEY_HIT_COUNT));
+ assertEquals(3L, gauge(metrics, KEY_MISS_COUNT));
+ }
+
+ /**
+ * A released counter must disappear rather than sit at zero, or a table
that performed no lookup at all
+ * would keep reporting the previous commit's values.
+ */
+ @Test
+ void releasingLeavesNoZeroValuedResidue() {
+ HoodieWriteConfig config = config(BASE_PATH, true);
+ Registry registry = seedRegistry(config, 10L, 7L, 3L);
+ HoodieMetrics metrics = metricsFor(config);
+
+ RecordIndexLookupMetrics.publishAndRelease(context, config, metrics);
+
+ assertTrue(registry.getAllCounts(false).isEmpty(),
+ "released counters must not linger, even at zero; got " +
registry.getAllCounts(false));
+ }
+
+ /**
+ * {@code Metrics.shutdown()} scrapes every registry in the process with
{@code flush=true}, so another
+ * table's write finishing can clear this registry mid-publish. The release
must clamp rather than
+ * subtract into negative numbers that every later commit would report.
+ */
+ @Test
+ void registryClearedBeforeReleaseDoesNotGoNegative() {
+ HoodieWriteConfig config = config(BASE_PATH, true);
+ Registry registry = seedRegistry(config, 10L, 7L, 3L);
+ HoodieMetrics metrics = metricsFor(config);
+
+ registry.clear();
+ RecordIndexLookupMetrics.publishAndRelease(context, config, metrics);
+
+ registry.getAllCounts(false).forEach((name, value) ->
+ assertTrue(value >= 0L, "release must clamp at zero, found " + name +
"=" + value));
+ }
+
+ /** Counts that arrive mid-publish belong to the next commit, and must
survive the release. */
+ @Test
+ void countsArrivingDuringThePublishSurviveTheRelease() {
+ HoodieWriteConfig config = config(BASE_PATH, true);
+ Registry registry = seedRegistry(config, 10L, 7L, 3L);
+ HoodieMetrics metrics = metricsFor(config);
+
+ // A straggler task's accumulator update, folded in before the drain reads.
+ registry.add(LOOKED_UP_KEY, 4L);
+ RecordIndexLookupMetrics.publishAndRelease(context, config, metrics);
+
+ assertTrue(registry.getAllCounts(false).isEmpty(),
+ "a drain that reads and releases the same values leaves nothing
behind");
+ assertEquals(14L, gauge(metrics, KEY_COUNT),
+ "the straggler's count is included in what was reported");
+ }
+
+ /** Two tables can share a name, so the registry key folds in the base path.
*/
+ @Test
+ void countersAreScopedByBasePathNotOnlyByTableName() {
+ HoodieWriteConfig config = config(BASE_PATH, true);
+ HoodieWriteConfig otherTable = config(OTHER_BASE_PATH, true);
+ seedRegistry(config, 10L, 7L, 3L);
+ HoodieMetrics metrics = metricsFor(otherTable);
+
+ RecordIndexLookupMetrics.publishAndRelease(context, otherTable, metrics);
+
+ assertNull(gauge(metrics, KEY_COUNT),
+ "a table at a different base path must not report another table's
lookups");
+ }
+
+ /** With the gate off, nothing is read and nothing is reported. */
+ @Test
+ void theGateSuppressesTheDrainEntirely() {
+ HoodieWriteConfig gatedOff = config(BASE_PATH, false);
+ Registry registry = seedRegistry(gatedOff, 10L, 7L, 3L);
+ HoodieMetrics metrics = metricsFor(gatedOff);
+
+ RecordIndexLookupMetrics.publishAndRelease(context, gatedOff, metrics);
+
+ assertNull(gauge(metrics, KEY_COUNT), "gate off: nothing reaches the
reporter");
+ assertFalse(registry.getAllCounts(false).isEmpty(),
+ "gate off: the registry is not consumed either");
+ }
+}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/config/metrics/HoodieMetricsConfig.java
b/hudi-common/src/main/java/org/apache/hudi/common/config/metrics/HoodieMetricsConfig.java
index e76017f128df..ea2beb8405fc 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/config/metrics/HoodieMetricsConfig.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/config/metrics/HoodieMetricsConfig.java
@@ -109,6 +109,21 @@ public class HoodieMetricsConfig extends HoodieConfig {
.sinceVersion("0.13.0")
.withDocumentation("Enable metrics for locking infra. Useful when
operating in multiwriter mode");
+ public static final ConfigProperty<Boolean> RLI_LOOKUP_METRICS_ENABLE =
ConfigProperty
+ .key(METRIC_PREFIX + ".rli.lookup.enable")
+ .defaultValue(false)
+ .markAdvanced()
+ .sinceVersion("1.3.0")
+ .withDocumentation("Collect counters for the record level index lookup
phase (records looked up, "
+ + "hits, misses, shards read and time spent) and report them at each
commit through the "
+ + "configured metrics reporter. Off unless set explicitly, including
when " + TURN_METRICS_ON.key()
+ + " is on: collection registers a Spark accumulator per table and
runs on every shard read, so "
+ + "it is opted into by name rather than inherited on upgrade.
Requires " + TURN_METRICS_ON.key()
+ + " as well, since the reporter is the only destination. Spark only:
the record level index "
+ + "lookup is a Spark write-path concern, and other engines produce
no counters. Counters are "
+ + "aggregated per table per JVM rather than per writer, so two write
clients on one table in "
+ + "one process share them.");
+
public static final ConfigProperty<String>
METRICS_REPORTER_FILE_BASED_CONFIGS_PATH = ConfigProperty
.key(METRIC_PREFIX + ".configs.properties")
.defaultValue("")
@@ -202,6 +217,10 @@ public class HoodieMetricsConfig extends HoodieConfig {
return getBoolean(HoodieMetricsConfig.LOCK_METRICS_ENABLE);
}
+ public boolean isRecordIndexLookupMetricsEnabled() {
+ return getBoolean(HoodieMetricsConfig.RLI_LOOKUP_METRICS_ENABLE);
+ }
+
public MetricsReporterType getMetricsReporterType() {
return
MetricsReporterType.valueOf(getString(HoodieMetricsConfig.METRICS_REPORTER_TYPE_VALUE));
}
@@ -411,6 +430,11 @@ public class HoodieMetricsConfig extends HoodieConfig {
return this;
}
+ public Builder withRecordIndexLookupMetrics(boolean enable) {
+ hoodieMetricsConfig.setValue(RLI_LOOKUP_METRICS_ENABLE,
String.valueOf(enable));
+ return this;
+ }
+
public Builder withLockingMetrics(boolean enable) {
hoodieMetricsConfig.setValue(LOCK_METRICS_ENABLE,
String.valueOf(enable));
return this;
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/config/metrics/TestHoodieMetricsConfig.java
b/hudi-common/src/test/java/org/apache/hudi/common/config/metrics/TestHoodieMetricsConfig.java
index f8a451f24923..daffe49c500c 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/config/metrics/TestHoodieMetricsConfig.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/config/metrics/TestHoodieMetricsConfig.java
@@ -22,6 +22,7 @@ import org.junit.jupiter.api.Test;
import static
org.apache.hudi.common.config.HoodieCommonConfig.META_SYNC_BASE_PATH_KEY;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
class TestHoodieMetricsConfig {
@@ -40,6 +41,18 @@ class TestHoodieMetricsConfig {
assertEquals("base/path/set/during/sync", config.getBasePath());
}
+ /**
+ * Turning metrics on is consent to report, not consent to start collecting
a new class of metric.
+ * Record index lookup collection registers a Spark accumulator per table
and runs per shard read, so
+ * an operator has to ask for it by name rather than inherit it on upgrade.
+ */
+ @Test
+ void recordIndexLookupMetricsStayOffUnlessAskedForByName() {
+ HoodieMetricsConfig config =
HoodieMetricsConfig.newBuilder().on(true).build();
+
assertFalse(config.getBoolean(HoodieMetricsConfig.RLI_LOOKUP_METRICS_ENABLE),
+ "enabling metrics must not silently enrol a table in record index
lookup collection");
+ }
+
@Test
void testReturnsNullWhenNeitherBasePathNorMetaSyncIsSet() {
HoodieMetricsConfig config = HoodieMetricsConfig.newBuilder().build();
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java
b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java
index 728f6dd6efc9..75bfd11277a9 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java
@@ -274,30 +274,6 @@ public class DataSourceUtils {
return false;
}
- /**
- * Resolves duplicate records in the provided {@code incomingHoodieRecords}.
- *
- * <p>If {@code failOnDuplicates} is {@code false}, duplicate records
already present in the dataset
- * are dropped. Otherwise, a {@link HoodieDuplicateKeyException} is thrown
if duplicates are found.</p>
- *
- * @param jssc the Spark context used for executing the deduplication
- * @param incomingHoodieRecords the input {@link JavaRDD} of {@link
HoodieRecord} objects to process
- * @param parameters a map of configuration parameters, including the
dataset path under the key {@code "path"}
- * @param failOnDuplicates a flag indicating whether to fail when duplicates
are found
- * @return a {@link JavaRDD} of deduplicated {@link HoodieRecord} objects
- */
- @SuppressWarnings("unchecked")
- public static JavaRDD<HoodieRecord> resolveDuplicates(JavaSparkContext jssc,
- JavaRDD<HoodieRecord>
incomingHoodieRecords,
- Map<String, String>
parameters,
- boolean
failOnDuplicates) {
- HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder()
- .withPath(parameters.get("path"))
- .withProps(parameters).build();
- return handleDuplicates(
- new HoodieSparkEngineContext(jssc), incomingHoodieRecords,
writeConfig, failOnDuplicates);
- }
-
/**
* Spark data source WriteStatus validator.
*
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
index 042af12a2954..46601c240493 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
@@ -550,7 +550,7 @@ class HoodieSparkSqlWriterInternal {
}
// Remove duplicates from incoming records based on existing keys
from storage.
- val dedupedHoodieRecords = handleInsertDuplicates(hoodieRecords,
hoodieConfig, operation, jsc, parameters)
+ val dedupedHoodieRecords = handleInsertDuplicates(hoodieRecords,
hoodieConfig, operation, client)
try {
val writeResult = DataSourceUtils.doWriteOperation(client,
dedupedHoodieRecords, instantTime, operation,
preppedSparkSqlWrites || preppedWriteOperation)
@@ -1234,15 +1234,18 @@ object HoodieSparkSqlWriterInternal {
def handleInsertDuplicates(incomingRecords: JavaRDD[HoodieRecord[_]],
hoodieConfig: HoodieConfig,
operation: WriteOperationType,
- jsc: JavaSparkContext,
- parameters: Map[String, String]):
JavaRDD[HoodieRecord[_]] = {
+ client: SparkRDDWriteClient[_]):
JavaRDD[HoodieRecord[_]] = {
// If no deduplication is needed, return the incoming records as is
if (!isDeduplicationRequired(hoodieConfig) ||
!isDeduplicationNeeded(operation)) {
incomingRecords
} else {
- // Perform deduplication
- DataSourceUtils.resolveDuplicates(
- jsc, incomingRecords, parameters.asJava,
shouldFailWhenDuplicatesFound(hoodieConfig))
+ // Resolve duplicates on the committing client's engine context and
config rather than a throwaway
+ // context. The record index lookup this triggers collects its counters
into a registry owned by the
+ // context that ran the lookup, and postCommit drains the client's
context; a separate context would
+ // strand those counters and the INSERT would publish none. See
RecordIndexLookupMetrics.
+ DataSourceUtils.handleDuplicates(
+ client.getEngineContext.asInstanceOf[HoodieSparkEngineContext],
incomingRecords,
+ client.getConfig, shouldFailWhenDuplicatesFound(hoodieConfig))
}
}
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionedRecordLevelIndexSupport.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionedRecordLevelIndexSupport.scala
index d2fe4589c095..cf75a09953ba 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionedRecordLevelIndexSupport.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionedRecordLevelIndexSupport.scala
@@ -103,6 +103,7 @@ class PartitionedRecordLevelIndexSupport(spark:
SparkSession,
.map(_._2)
.toJavaRDD()
ValidationUtils.checkState(partitionedKeyRDD.getNumPartitions <=
numFileGroups)
+ // Read path: no write-side registry to report into, so lookups here are
not instrumented.
val fileIdToPartitionMap = partitionedKeyRDD.mapPartitionsToPair(new
PartitionedRecordIndexFileGroupLookupFunction(metadataTable))
.collect()
.asScala
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/TestDataSourceUtils.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/TestDataSourceUtils.java
index 8755120bb641..d8c90da5e7d0 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/TestDataSourceUtils.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/TestDataSourceUtils.java
@@ -66,7 +66,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -341,7 +340,6 @@ public class TestDataSourceUtils extends
HoodieClientTestBase {
void testDeduplicationAgainstRecordsAlreadyInTable() throws IOException {
initResources();
HoodieWriteConfig config = getConfig();
- config.getProps().setProperty("path", config.getBasePath());
try (SparkRDDWriteClient writeClient = getHoodieWriteClient(config)) {
String newCommitTime = writeClient.startCommit();
List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 100);
@@ -350,10 +348,11 @@ public class TestDataSourceUtils extends
HoodieClientTestBase {
writeClient.commit(newCommitTime, jsc.parallelize(statusList),
Option.empty(), COMMIT_ACTION, Collections.emptyMap(), Option.empty());
assertNoWriteErrors(statusList);
- Map<String, String> parameters =
config.getProps().entrySet().stream().collect(Collectors.toMap(entry ->
entry.getKey().toString(), entry -> entry.getValue().toString()));
List<HoodieRecord> newRecords = dataGen.generateInserts(newCommitTime,
10);
List<HoodieRecord> inputRecords = Stream.concat(records.subList(0,
10).stream(), newRecords.stream()).collect(Collectors.toList());
- List<HoodieRecord> output = DataSourceUtils.resolveDuplicates(jsc,
jsc.parallelize(inputRecords, 1), parameters, false).collect();
+ // Deduplicate against the committing client's engine context and
config, the same wiring the
+ // Spark SQL writer uses so the record index lookup registry is owned by
the draining context.
+ List<HoodieRecord> output = DataSourceUtils.handleDuplicates(context,
jsc.parallelize(inputRecords, 1), config, false).collect();
Set<String> expectedRecordKeys =
newRecords.stream().map(HoodieRecord::getRecordKey).collect(Collectors.toSet());
assertEquals(expectedRecordKeys,
output.stream().map(HoodieRecord::getRecordKey).collect(Collectors.toSet()));
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/testutils/CapturingMetricsReporter.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/testutils/CapturingMetricsReporter.java
new file mode 100644
index 000000000000..bb10617d27fb
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/testutils/CapturingMetricsReporter.java
@@ -0,0 +1,79 @@
+/*
+ * 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.testutils;
+
+import org.apache.hudi.metrics.custom.CustomizableMetricsReporter;
+
+import com.codahale.metrics.MetricRegistry;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/**
+ * Records every gauge value handed to a reporter, so a test can read them
after the write.
+ *
+ * <p>Two write paths behave differently and both have to be readable. The
Spark DataSource shuts metrics
+ * down at the end of every write ({@code DefaultSource}), which reports once
and then discards the
+ * registry, so the value has to be captured as it is reported or it is gone.
Spark SQL does not shut
+ * down, so nothing triggers a report at all and the value sits in a
still-live registry. {@link
+ * #captured()} therefore polls every reporter still attached to a registry
before answering.
+ */
+public class CapturingMetricsReporter extends CustomizableMetricsReporter {
+
+ private static final Map<String, Long> CAPTURED = new ConcurrentHashMap<>();
+ /** Reporters whose registry is still live, so a not-yet-reported gauge can
still be read. */
+ private static final List<CapturingMetricsReporter> ATTACHED = new
CopyOnWriteArrayList<>();
+
+ public CapturingMetricsReporter(Properties props, MetricRegistry registry) {
+ super(props, registry);
+ ATTACHED.add(this);
+ }
+
+ public static Map<String, Long> captured() {
+ ATTACHED.forEach(CapturingMetricsReporter::report);
+ return CAPTURED;
+ }
+
+ public static void reset() {
+ CAPTURED.clear();
+ }
+
+ @Override
+ public void start() {
+ }
+
+ @Override
+ public void report() {
+ getRegistry().getGauges().forEach((name, gauge) -> {
+ Object value = gauge.getValue();
+ if (value instanceof Number) {
+ CAPTURED.put(name, ((Number) value).longValue());
+ }
+ });
+ }
+
+ @Override
+ public void stop() {
+ // Detach before the registry goes stale, so a later reset() is not undone
by re-reading dead gauges.
+ ATTACHED.remove(this);
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/RliLookupMetricsTestBase.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/RliLookupMetricsTestBase.scala
new file mode 100644
index 000000000000..addaae8254eb
--- /dev/null
+++
b/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.config.metrics.HoodieMetricsConfig
+import org.apache.hudi.config.HoodieIndexConfig
+import org.apache.hudi.metrics.{MetricsReporterType, RecordIndexLookupMetrics}
+import org.apache.hudi.testutils.CapturingMetricsReporter
+
+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 configured metrics
reporter.
+ */
+abstract class RliLookupMetricsTestBase extends RecordLevelIndexTestBase {
+
+ /** Overridden by the partitioned subclasses; both variants are separate
closures on separate paths. */
+ protected def isPartitionedRli: Boolean = false
+
+ /**
+ * Copy-on-write throughout. Tagging is an index-level concern and nothing
in the lookup closures or the
+ * drain branches on table type, so merge-on-read exercises identical code
and is not covered separately.
+ */
+ protected def indexLabel: String = if (isPartitionedRli) "partitioned RLI"
else "global RLI"
+
+ /**
+ * The drain reports to the configured reporter, so these tests need one.
`commonOpts` turns the global
+ * record index on, so the metadata-partition flags and the index type are
flipped together below to
+ * select the partitioned variant.
+ */
+ protected def metricsOpts: Map[String, String] = Map(
+ HoodieMetricsConfig.TURN_METRICS_ON.key -> "true",
+ // Explicit: collection is opted into by name, not inherited from metrics
being on.
+ HoodieMetricsConfig.RLI_LOOKUP_METRICS_ENABLE.key -> "true",
+ // The type still has to be set: it defaults to GRAPHITE, whose config
builder NPEs without a prefix.
+ // The factory prefers the class when one is given, so this only keeps the
builder happy.
+ HoodieMetricsConfig.METRICS_REPORTER_TYPE_VALUE.key ->
MetricsReporterType.INMEMORY.name(),
+ HoodieMetricsConfig.METRICS_REPORTER_CLASS_NAME.key ->
classOf[CapturingMetricsReporter].getName)
+
+ protected def rliOpts: Map[String, String] = {
+ val withTableType = Map(DataSourceWriteOptions.TABLE_TYPE.key ->
+ DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL) ++ metricsOpts
+ if (isPartitionedRli) {
+ commonOpts ++ withTableType ++ Map(
+ HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key ->
"false",
+ HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key -> "true",
+ HoodieIndexConfig.INDEX_TYPE.key -> "RECORD_LEVEL_INDEX")
+ } else {
+ commonOpts ++ withTableType ++ Map(
+ HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key ->
"true",
+ HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key -> "false",
+ HoodieIndexConfig.INDEX_TYPE.key -> "GLOBAL_RECORD_LEVEL_INDEX")
+ }
+ }
+
+ /**
+ * A lookup that happened stamps the full counter set, zeros included, so an
absent key means nothing was
+ * looked up at all. The default is defensive against exactly that case.
+ */
+ protected def counterOrZero(counters: Map[String, String], metric: String):
Long =
+ counters.getOrElse(metric, "0").toLong
+
+ /**
+ * The counters as an operator would read them: off the metrics reporter.
{@code Metrics} is keyed by
+ * base path, so building a handle here returns the same instance the write
published into.
+ */
+ protected def rliCountersFromLatestCommit(): Map[String, String] = {
+ val marker = "." + RecordIndexLookupMetrics.METRIC_ACTION +
+ "." + RecordIndexLookupMetrics.METRIC_QUALIFIER + "."
+ CapturingMetricsReporter.captured().asScala.toMap
+ .collect { case (name, value) if name.contains(marker) =>
+ name.substring(name.indexOf(marker) + marker.length) -> value.toString
}
+ }
+
+ /** The reporter is process-wide, so each test starts from a clean slate. */
+ @org.junit.jupiter.api.BeforeEach
+ def resetCapturedMetrics(): Unit = CapturingMetricsReporter.reset()
+
+
+ /** Asserts the core invariant and returns the looked-up count. */
+ protected def assertSumInvariant(counters: Map[String, String]): Long = {
+ val lookedUp = counterOrZero(counters, RecordIndexLookupMetrics.KEY_COUNT)
+ val hits = counterOrZero(counters, RecordIndexLookupMetrics.KEY_HIT_COUNT)
+ val misses = counterOrZero(counters,
RecordIndexLookupMetrics.KEY_MISS_COUNT)
+ org.junit.jupiter.api.Assertions.assertEquals(lookedUp, hits + misses,
+ "hits + misses must account for every key looked up")
+ // A lookup that happened must also report the time it took, or the timing
metric is silently absent on
+ // paths nobody checked. Zero is allowed: a shard read can round below a
millisecond.
+ if (lookedUp > 0) {
+ org.junit.jupiter.api.Assertions.assertTrue(
+ counters.contains(RecordIndexLookupMetrics.LOOKUP_TIME),
+ s"looked up $lookedUp keys but reported no
${RecordIndexLookupMetrics.LOOKUP_TIME}; " +
+ s"counters were ${counters.keys.toSeq.sorted.mkString(", ")}")
+ }
+ lookedUp
+ }
+
+ /** Everything on the latest commit, for diagnosing an unexpectedly empty
counter set. */
+ protected def allExtraMetadataFromLatestCommit(): Map[String, String] = {
+ metaClient.reloadActiveTimeline()
+ val lastInstant =
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants().lastInstant().get()
+
metaClient.getActiveTimeline.readCommitMetadata(lastInstant).getExtraMetadata.asScala.toMap
+
+ ("__instant" -> lastInstant.toString)
+ }
+
+ protected def report(label: String, counters: Map[String, String]): Unit = {
+ println(s"\n===== $label =====")
+ if (counters.isEmpty) {
+ println(" (no RLI counters published) -- every gauge the reporter holds
follows:")
+ CapturingMetricsReporter.captured().asScala.toSeq.sortBy(_._1).foreach {
case (k, v) =>
+ println(f" $k%-70s $v")
+ }
+ } else {
+ counters.toSeq.sorted.foreach { case (k, v) => println(f" $k%-52s $v") }
+ }
+ println("=" * (12 + label.length) + "\n")
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestRliLookupMetricsAcrossFailedCommit.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestRliLookupMetricsAcrossFailedCommit.scala
new file mode 100644
index 000000000000..c5ae79386f3a
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestRliLookupMetricsAcrossFailedCommit.scala
@@ -0,0 +1,125 @@
+/*
+ * 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.client.transaction.SimpleConcurrentFileWritesConflictResolutionStrategy
+import org.apache.hudi.common.model.WriteConcurrencyMode
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.HoodieInstant
+import org.apache.hudi.common.util.{Option => HoodieOption}
+import org.apache.hudi.config.{HoodieLockConfig, HoodieWriteConfig}
+import org.apache.hudi.exception.HoodieWriteConflictException
+import org.apache.hudi.metrics.RecordIndexLookupMetrics
+import org.apache.hudi.testutils.CapturingMetricsReporter
+
+import org.apache.spark.sql.SaveMode
+import org.junit.jupiter.api.{Tag, Test}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+
+/** A commit that never lands must not take its lookup counters with it. */
+@Tag("functional")
+class TestRliLookupMetricsAcrossFailedCommit extends RliLookupMetricsTestBase {
+
+ /** Options that make `preCommit` throw, i.e. after the snapshot and before
the commit completes. */
+ private def conflictingOpts: Map[String, String] = rliOpts ++ Map(
+ HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key ->
WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL.name,
+ HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key ->
+ "org.apache.hudi.client.transaction.lock.InProcessLockProvider",
+ HoodieLockConfig.WRITE_CONFLICT_RESOLUTION_STRATEGY_CLASS_NAME.key ->
+ classOf[AlwaysConflictingResolutionStrategy].getName)
+
+ private def causeChain(t: Throwable): String = {
+ var current = t
+ val sb = new StringBuilder
+ while (current != null) {
+ sb.append(current.toString).append(" | ")
+ current = current.getCause
+ }
+ sb.toString
+ }
+
+ @Test
+ def testCountersSurviveACommitThatNeverLands(): Unit = {
+ val failedUpdates = 10
+ val retriedUpdates = 4
+ // Each upsert batch carries one fresh insert alongside its updates, so it
looks up N + 1 keys.
+ val retriedLookups = retriedUpdates + 1
+
+ doWriteAndValidateDataAndRecordIndex(rliOpts, INSERT_OPERATION_OPT_VAL,
SaveMode.Overwrite,
+ validate = false, numInserts = 80)
+ assertTrue(rliCountersFromLatestCommit().isEmpty, "the seeding insert
performs no lookup")
+
+ // An upsert whose commit is rejected in preCommit. The lookups happened;
the commit did not.
+ val failure = try {
+ doWriteAndValidateDataAndRecordIndex(conflictingOpts,
UPSERT_OPERATION_OPT_VAL, SaveMode.Append,
+ validate = false, numUpdates = failedUpdates)
+ None
+ } catch {
+ case t: Throwable => Some(t)
+ }
+ assertTrue(failure.isDefined, "the injected conflict must fail the write")
+
assertTrue(causeChain(failure.get).contains(AlwaysConflictingResolutionStrategy.MESSAGE),
+ s"the write must have failed on the injected conflict, not on something
else: ${causeChain(failure.get)}")
+
+ // Nothing was published for it: the newest completed commit is still the
seeding insert.
+ val afterFailure = rliCountersFromLatestCommit()
+ report(s"Failed commit ($indexLabel) -- latest completed commit, expected
empty", afterFailure)
+ assertTrue(afterFailure.isEmpty,
+ s"a commit that never landed must not leave counters on the timeline;
got $afterFailure")
+
+ // The next commit to succeed reports normally, and the failed attempt's
counters do not corrupt it.
+ CapturingMetricsReporter.reset()
+ doWriteAndValidateDataAndRecordIndex(rliOpts, UPSERT_OPERATION_OPT_VAL,
SaveMode.Append,
+ validate = false, numUpdates = retriedUpdates)
+
+ val counters = rliCountersFromLatestCommit()
+ report(s"Retry after failed commit ($indexLabel) -- expected
$retriedLookups", counters)
+
+ assertTrue(counters.nonEmpty, "the retry must carry counters")
+ // The failed attempt's lookups are not carried into this commit.
Publishing is skipped when the
+ // commit does not land, but the registry does not survive to the retry
either: the DataSource path
+ // tears metrics down after every write, and Metrics.shutdown()
flush-and-clears every Registry. So on
+ // this path an abandoned attempt's counters are dropped rather than
double-counted, which is the safer
+ // of the two failure modes but is not a carry-forward guarantee.
+ assertEquals(retriedLookups.toLong,
+ assertSumInvariant(counters),
+ "the retry must report exactly its own lookups, uncontaminated by the
attempt that never landed")
+ }
+}
+
+object AlwaysConflictingResolutionStrategy {
+ val MESSAGE = "injected conflict: this commit must not land"
+}
+
+/**
+ * Fails conflict resolution unconditionally, which is the first thing
`preCommit` does. Loaded reflectively
+ * from `hoodie.write.lock.conflict.resolution.strategy`, so it needs a
no-argument constructor.
+ */
+class AlwaysConflictingResolutionStrategy extends
SimpleConcurrentFileWritesConflictResolutionStrategy {
+ override def getCandidateInstants(metaClient: HoodieTableMetaClient,
+ currentInstant: HoodieInstant,
+ lastSuccessfulInstant:
HoodieOption[HoodieInstant]): java.util.stream.Stream[HoodieInstant] =
+ throw new
HoodieWriteConflictException(AlwaysConflictingResolutionStrategy.MESSAGE)
+}
+
+/** The same coverage against the partitioned record level index. */
+@Tag("functional")
+class TestRliLookupMetricsAcrossFailedCommitPartitioned extends
TestRliLookupMetricsAcrossFailedCommit {
+ override protected def isPartitionedRli: Boolean = true
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestRliLookupMetricsOnDataSource.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestRliLookupMetricsOnDataSource.scala
new file mode 100644
index 000000000000..f4bf0b4b0301
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestRliLookupMetricsOnDataSource.scala
@@ -0,0 +1,154 @@
+/*
+ * 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.testutils.HoodieTestDataGenerator
+import
org.apache.hudi.common.testutils.HoodieTestDataGenerator.recordsToStrings
+import org.apache.hudi.metrics.RecordIndexLookupMetrics
+import org.apache.hudi.testutils.CapturingMetricsReporter
+
+import org.apache.spark.sql.SaveMode
+import org.junit.jupiter.api.{Tag, Test}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+
+import scala.collection.JavaConverters._
+
+/** Record level index lookup counters on the Spark DataSource write path. */
+@Tag("functional")
+class TestRliLookupMetricsOnDataSource extends RliLookupMetricsTestBase {
+
+ /**
+ * An upsert of N updates also carries exactly one fresh insert (see
+ * `RecordLevelIndexTestBase.doWriteAndValidateDataAndRecordIndex`), so the
miss count is 1.
+ */
+ @Test
+ def testCountersReachTheReporter(): Unit = {
+ val numUpdates = 20
+
+ doWriteAndValidateDataAndRecordIndex(rliOpts, INSERT_OPERATION_OPT_VAL,
SaveMode.Overwrite, validate = false, numInserts = 100)
+
+ doWriteAndValidateDataAndRecordIndex(rliOpts, UPSERT_OPERATION_OPT_VAL,
SaveMode.Append, validate = false, numUpdates = numUpdates)
+
+ val counters = rliCountersFromLatestCommit()
+ report(s"DataSource ($indexLabel) -- RLI counters on the commit", counters)
+
+ assertTrue(counters.nonEmpty, "the reporter must carry the RLI counters")
+ assertEquals(numUpdates.toString,
counters(RecordIndexLookupMetrics.KEY_HIT_COUNT), "every updated key is a hit")
+ assertEquals("1", counters(RecordIndexLookupMetrics.KEY_MISS_COUNT), "the
fresh insert is a miss")
+ assertEquals(numUpdates + 1L, assertSumInvariant(counters))
+ assertTrue(counters(RecordIndexLookupMetrics.SHARDS_READ).toInt > 0, "at
least one shard was read")
+ }
+
+ /** Each commit must report only its own work: the drain clears the registry
as it publishes. */
+ @Test
+ def testCountersArePerCommitNotCumulative(): Unit = {
+ val numUpdates = 10
+ doWriteAndValidateDataAndRecordIndex(rliOpts, INSERT_OPERATION_OPT_VAL,
SaveMode.Overwrite, validate = false, numInserts = 100)
+
+ val perCommit = (1 to 3).map { commit =>
+ doWriteAndValidateDataAndRecordIndex(rliOpts, UPSERT_OPERATION_OPT_VAL,
SaveMode.Append, validate = false, numUpdates = numUpdates)
+ val lookedUp =
rliCountersFromLatestCommit().getOrElse(RecordIndexLookupMetrics.KEY_COUNT, "0")
+ println(s"[per-commit] commit $commit ($indexLabel):
records_looked_up=$lookedUp")
+ lookedUp
+ }
+
+ perCommit.zipWithIndex.foreach { case (v, i) =>
+ assertEquals((numUpdates + 1).toString, v,
+ s"commit ${i + 1} must report only its own ${numUpdates + 1} lookups,
not a running total")
+ }
+ }
+
+ /** A commit that performed no lookup must publish nothing of its own. */
+ @Test
+ def testACommitWithNoLookupCarriesNoCounters(): Unit = {
+ doWriteAndValidateDataAndRecordIndex(rliOpts, INSERT_OPERATION_OPT_VAL,
SaveMode.Overwrite,
+ validate = false, numInserts = 60)
+
+ // A commit that does tag, so the registry is drained for the first time.
+ doWriteAndValidateDataAndRecordIndex(rliOpts, UPSERT_OPERATION_OPT_VAL,
SaveMode.Append,
+ validate = false, numUpdates = 10)
+ assertTrue(rliCountersFromLatestCommit().nonEmpty, "the upsert must
publish counters to drain")
+
+ // The reporter is a stream of emissions, not a running record, so clear
what the upsert emitted
+ // before the write under test. What remains afterwards is what this
commit alone published.
+ CapturingMetricsReporter.reset()
+
+ // A plain insert performs no index lookup, so its commit has nothing to
report.
+ doWriteAndValidateDataAndRecordIndex(rliOpts, INSERT_OPERATION_OPT_VAL,
SaveMode.Append,
+ validate = false, numInserts = 5)
+
+ val counters = rliCountersFromLatestCommit()
+ report(s"DataSource ($indexLabel) -- insert after a drained upsert,
expecting nothing", counters)
+ assertTrue(counters.isEmpty,
+ s"a commit that looked nothing up must publish no counters of its own;
got $counters")
+ }
+
+ /**
+ * INSERT with drop-dups resolves duplicates through a record index lookup
before the write, and that
+ * lookup runs on the committing client's engine context. Its counters must
therefore reach the reporter
+ * at the INSERT's commit. Regression guard: dedup used to run on a
throwaway context (built inside
+ * `DataSourceUtils`) whose registry `postCommit` never drained, so the
INSERT published nothing.
+ */
+ @Test
+ def testInsertDropDupsPublishesDedupLookupCounters(): Unit = {
+ val numSeed = 100
+ val numReoffered = 40 // existing keys offered again -> dropped as
duplicates -> hits
+ val numFresh = 15 // brand-new keys -> kept -> misses
+
+ // Seed the table so the record index has keys to hit.
+ val seedBatch = doWriteAndValidateDataAndRecordIndex(rliOpts,
INSERT_OPERATION_OPT_VAL, SaveMode.Overwrite,
+ validate = false, numInserts = numSeed)
+
+ // A batch mixing already-present keys with new ones, so both hits and
misses are exercised.
+ val freshBatch = recordsToStrings(dataGen.generateInsertsAsPerSchema(
+ getInstantTime(), numFresh,
HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA)).asScala
+ val freshDf =
spark.read.json(spark.sparkContext.parallelize(freshBatch.toSeq, 2))
+ val insertBatch = seedBatch.limit(numReoffered).unionByName(freshDf)
+
+ // Isolate the INSERT under test from the seed's emissions.
+ CapturingMetricsReporter.reset()
+
+ insertBatch.write.format("hudi")
+ .options(rliOpts)
+ .option(OPERATION.key, INSERT_OPERATION_OPT_VAL)
+ .option(INSERT_DROP_DUPS.key, "true")
+ .mode(SaveMode.Append)
+ .save(basePath)
+
+ val counters = rliCountersFromLatestCommit()
+ report(s"DataSource INSERT drop-dups ($indexLabel)", counters)
+
+ assertTrue(counters.nonEmpty,
+ "INSERT with drop-dups resolves duplicates via an RLI lookup; its
counters must reach the reporter")
+ assertEquals(numReoffered.toString,
counters(RecordIndexLookupMetrics.KEY_HIT_COUNT),
+ "the re-offered keys already exist in the index")
+ assertEquals(numFresh.toString,
counters(RecordIndexLookupMetrics.KEY_MISS_COUNT),
+ "the brand-new keys are misses")
+ assertEquals((numReoffered + numFresh).toLong,
assertSumInvariant(counters),
+ "the dedup lookup examines every incoming record")
+ assertTrue(counters(RecordIndexLookupMetrics.SHARDS_READ).toInt > 0, "at
least one shard was read")
+ }
+
+}
+
+/** The same coverage against the partitioned record level index. */
+@Tag("functional")
+class TestRliLookupMetricsOnDataSourcePartitioned extends
TestRliLookupMetricsOnDataSource {
+ override protected def isPartitionedRli: Boolean = true
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestRliLookupMetricsOnSparkSql.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestRliLookupMetricsOnSparkSql.scala
new file mode 100644
index 000000000000..b4cd09e8eac9
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestRliLookupMetricsOnSparkSql.scala
@@ -0,0 +1,138 @@
+/*
+ * 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.RecordIndexLookupMetrics
+import org.apache.hudi.testutils.CapturingMetricsReporter
+
+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"))
+
+ // SQL DML does not inherit the DataSource options the seed write used, so
the reporter has to be
+ // selected as a session config too or the write publishes nowhere the
test can read.
+ metricsOpts.foreach { case (key, value) => spark.sql(s"set $key = $value")
}
+ CapturingMetricsReporter.reset()
+ }
+
+ /**
+ * 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,
+ "a prepped UPDATE already knows each record's location, so it performs
no RLI lookup to report")
+ }
+
+ /**
+ * With optimized writes disabled the UPDATE goes through normal tagging,
and every touched key is
+ * looked up in the index. No predicate means every row, so the counts are
exact.
+ */
+ @Test
+ def testUpdateWithoutOptimizedWritesPublishesCounters(): Unit = {
+ seedTableAndRegisterForSql()
+ spark.sql(s"set ${SPARK_SQL_OPTIMIZED_WRITES.key} = false")
+
+ spark.sql(s"update $sqlTable set rider = 'rider-updated'")
+
+ val counters = rliCountersFromLatestCommit()
+ report(s"Spark SQL UPDATE, optimized writes OFF ($indexLabel)", counters)
+
+ assertTrue(counters.nonEmpty, "a non-prepped UPDATE must publish RLI
counters at its commit")
+ val lookedUp = assertSumInvariant(counters)
+ assertEquals(numSeedRecords.toLong, lookedUp, "UPDATE with no predicate
tags every row")
+ assertEquals(numSeedRecords.toString,
counters(RecordIndexLookupMetrics.KEY_HIT_COUNT),
+ "every row being updated already exists in the index")
+ // A caller that looked something up stamps its whole counter set, zeros
included, so the record on the
+ // timeline is internally consistent. Dropping just the zero components
would leave a commit reporting
+ // hits and records_looked_up but no misses, forcing every consumer to
treat absent as zero.
+ assertTrue(counters.contains(RecordIndexLookupMetrics.KEY_MISS_COUNT),
+ "a caller that looked keys up reports its full counter set, including
the zeros")
+ assertEquals("0", counters(RecordIndexLookupMetrics.KEY_MISS_COUNT), "no
new keys are introduced")
+ assertTrue(counters(RecordIndexLookupMetrics.SHARDS_READ).toInt > 0, "at
least one shard was read")
+ }
+
+ /** MERGE INTO is not a prepped write, so its keys are tagged and must be
counted. */
+ @Test
+ def testMergeIntoPublishesCounters(): Unit = {
+ seedTableAndRegisterForSql()
+
+ val numMerged = 25
+ spark.sql(
+ s"""create or replace temporary view rli_merge_src as
+ |select _row_key, partition, timestamp, 'rider-merged' as rider
+ |from $sqlTable limit $numMerged""".stripMargin)
+
+ spark.sql(
+ s"""merge into $sqlTable t
+ |using rli_merge_src s
+ |on t._row_key = s._row_key
+ |when matched then update set t.rider = s.rider, t.timestamp =
s.timestamp""".stripMargin)
+
+ val counters = rliCountersFromLatestCommit()
+ report(s"Spark SQL MERGE INTO ($indexLabel)", counters)
+
+ assertTrue(counters.nonEmpty, "a MERGE INTO must publish RLI counters at
its commit")
+ val lookedUp = assertSumInvariant(counters)
+ assertEquals(numMerged.toLong, lookedUp, s"the merge tags its $numMerged
matched keys")
+ assertEquals(numMerged.toString,
counters(RecordIndexLookupMetrics.KEY_HIT_COUNT),
+ "every merged key already exists in the index")
+ }
+}
+
+/** The same coverage against the partitioned record level index. */
+@Tag("functional")
+class TestRliLookupMetricsOnSparkSqlPartitioned extends
TestRliLookupMetricsOnSparkSql {
+ override protected def isPartitionedRli: Boolean = true
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestRliMetricsOnStreamerPath.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestRliMetricsOnStreamerPath.java
new file mode 100644
index 000000000000..1d83653f01c4
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestRliMetricsOnStreamerPath.java
@@ -0,0 +1,128 @@
+/*
+ * 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.utilities.deltastreamer;
+
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.metrics.HoodieMetricsConfig;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.metrics.RecordIndexLookupMetrics;
+import org.apache.hudi.utilities.testutils.CapturingMetricsReporter;
+
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The RLI lookup counters must reach commit metadata on the DeltaStreamer
path, not only on the Spark DataSource path.
+ */
+@Tag("functional")
+public class TestRliMetricsOnStreamerPath extends HoodieDeltaStreamerTestBase {
+
+ /** Selects the global or partitioned record index. */
+ private static void enableRecordIndex(HoodieDeltaStreamer.Config cfg,
boolean partitioned) {
+ cfg.configs.add(HoodieMetadataConfig.ENABLE.key() + "=true");
+ cfg.configs.add(HoodieMetricsConfig.TURN_METRICS_ON.key() + "=true");
+ cfg.configs.add(HoodieMetricsConfig.RLI_LOOKUP_METRICS_ENABLE.key() +
"=true");
+ cfg.configs.add(HoodieMetricsConfig.METRICS_REPORTER_TYPE_VALUE.key() +
"=INMEMORY");
+ cfg.configs.add(HoodieMetricsConfig.METRICS_REPORTER_CLASS_NAME.key() + "="
+ + CapturingMetricsReporter.class.getName());
+
cfg.configs.add(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key()
+ "=" + !partitioned);
+ cfg.configs.add(HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key()
+ "=" + partitioned);
+ cfg.configs.add(HoodieIndexConfig.INDEX_TYPE.key() + "="
+ + (partitioned ? "RECORD_LEVEL_INDEX" : "GLOBAL_RECORD_LEVEL_INDEX"));
+ }
+
+ /**
+ * Reads the counters the way an operator would, off the configured
reporter. Goes through the reporter
+ * rather than {@code Metrics}, which is keyed by base path and cannot be
addressed reliably from here.
+ */
+ private static Map<String, String> rliCountersOnLatestCommit(String
tableBasePath) {
+ String marker = RecordIndexLookupMetrics.METRIC_ACTION
+ + "." + RecordIndexLookupMetrics.METRIC_QUALIFIER + ".";
+ Map<String, String> rli = new HashMap<>();
+ CapturingMetricsReporter.captured().forEach((name, value) -> {
+ int at = name.indexOf(marker);
+ if (at == 0 || (at > 0 && name.charAt(at - 1) == '.')) {
+ rli.put(name.substring(at + marker.length()), String.valueOf(value));
+ }
+ });
+ return rli;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testRliCountersReachTheReporterOnStreamerPath(boolean
partitioned) throws Exception {
+ String label = partitioned ? "partitioned" : "global";
+ String tableBasePath = basePath + "/test_rli_metrics_streamer_" + label;
+
+ // Sync 1 -- build the table and the record index.
+ HoodieDeltaStreamer.Config insertCfg =
+ TestHoodieDeltaStreamer.TestHelpers.makeConfig(tableBasePath,
WriteOperationType.INSERT);
+ enableRecordIndex(insertCfg, partitioned);
+ new HoodieDeltaStreamer(insertCfg, jsc).sync();
+
+ // Sync 2 -- upsert, which tags incoming keys against the record index.
+ HoodieDeltaStreamer.Config upsertCfg =
+ TestHoodieDeltaStreamer.TestHelpers.makeConfig(tableBasePath,
WriteOperationType.UPSERT);
+ enableRecordIndex(upsertCfg, partitioned);
+ new HoodieDeltaStreamer(upsertCfg, jsc).sync();
+
+ Map<String, String> counters = rliCountersOnLatestCommit(tableBasePath);
+
+ System.out.println("\n===== DeltaStreamer (" + label + " RLI) -- RLI
counters on the commit =====");
+ if (counters.isEmpty()) {
+ System.out.println(" (none found)");
+ } else {
+ counters.entrySet().stream()
+ .sorted(Map.Entry.comparingByKey())
+ .forEach(e -> System.out.println(String.format(" %-52s %s",
e.getKey(), e.getValue())));
+ }
+
System.out.println("==========================================================\n");
+
+ assertFalse(counters.isEmpty(),
+ "the commit-boundary drain must fire on the DeltaStreamer path;
hudi-utilities never calls "
+ + "Metrics.shutdownAllMetrics, so nothing else would publish
these");
+
+ String lookedUp = RecordIndexLookupMetrics.KEY_COUNT;
+ assertTrue(counters.containsKey(lookedUp),
+ "tag-location traffic must be attributed on the streamer path too; got
" + counters.keySet());
+
+ long records = Long.parseLong(counters.get(lookedUp));
+ long hits =
Long.parseLong(counters.get(RecordIndexLookupMetrics.KEY_HIT_COUNT));
+ long misses =
Long.parseLong(counters.get(RecordIndexLookupMetrics.KEY_MISS_COUNT));
+ // Exact, not an invariant: misses is derived as records - hits at the
emission site, so
+ // records == hits + misses holds by construction and would survive a
doubled count. The workload is
+ // deterministic -- the first sync writes 1000 records, the second updates
500 and inserts 500.
+ assertEquals(1000L, records, "the upsert sync looked up every key from the
first sync");
+ assertEquals(500L, hits, "the 500 updates hit the index");
+ assertEquals(500L, misses, "the 500 fresh inserts missed");
+ // Shard count is not pinned: it follows the index file-group layout,
which differs between the
+ // global and partitioned variants (10 and 3 on this workload).
+
assertTrue(Long.parseLong(counters.get(RecordIndexLookupMetrics.SHARDS_READ)) >
0,
+ "at least one shard was read");
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/CapturingMetricsReporter.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/CapturingMetricsReporter.java
new file mode 100644
index 000000000000..fede84cab130
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/CapturingMetricsReporter.java
@@ -0,0 +1,73 @@
+/*
+ * 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.utilities.testutils;
+
+import org.apache.hudi.metrics.custom.CustomizableMetricsReporter;
+
+import com.codahale.metrics.MetricRegistry;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/**
+ * Records gauge values so a test can read them without having to guess which
{@code Metrics} instance the
+ * write published into -- that is keyed by base path, and reconstructing the
key from outside is fragile.
+ */
+public class CapturingMetricsReporter extends CustomizableMetricsReporter {
+
+ private static final Map<String, Long> CAPTURED = new ConcurrentHashMap<>();
+ private static final List<CapturingMetricsReporter> ATTACHED = new
CopyOnWriteArrayList<>();
+
+ public CapturingMetricsReporter(Properties props, MetricRegistry registry) {
+ super(props, registry);
+ ATTACHED.add(this);
+ }
+
+ /** Polls every still-attached registry first: nothing forces a report on
the DeltaStreamer path. */
+ public static Map<String, Long> captured() {
+ ATTACHED.forEach(CapturingMetricsReporter::report);
+ return CAPTURED;
+ }
+
+ public static void reset() {
+ CAPTURED.clear();
+ }
+
+ @Override
+ public void start() {
+ }
+
+ @Override
+ public void report() {
+ getRegistry().getGauges().forEach((name, gauge) -> {
+ Object value = gauge.getValue();
+ if (value instanceof Number) {
+ CAPTURED.put(name, ((Number) value).longValue());
+ }
+ });
+ }
+
+ @Override
+ public void stop() {
+ ATTACHED.remove(this);
+ }
+}
