This is an automated email from the ASF dual-hosted git repository.
danny0405 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 59ad99f4402f feat(flink): report the steps of a Hive sync in the write
coordinator metrics (#20165)
59ad99f4402f is described below
commit 59ad99f4402f570a89eb0bbe0270d1fc1b5f8cd0
Author: Shihuan Liu <[email protected]>
AuthorDate: Thu Oct 22 18:57:32 2026 -0700
feat(flink): report the steps of a Hive sync in the write coordinator
metrics (#20165)
* feat(flink): report the steps of a Hive sync in the write coordinator
metrics
The Hive sync metrics on the write coordinator say whether a sync
succeeded and how long it took, but not where that time went or what
the sync changed. HiveSyncTool now records per sync what it did, in a
HiveSyncStats that callers read through getSyncStats(): time reading
the schema from storage, time finding partitions to sync, the remaining
metastore time, partitions added, and whether the schema evolved.
The coordinator reports them next to the existing metrics:
- hiveSyncInitDurationMs (histogram, building the sync tool)
- hiveSyncSchemaReadDurationMs (histogram)
- hiveSyncPartitionScanDurationMs (histogram)
- hiveSyncMetastoreDurationMs (histogram)
- hiveSyncPartitionsAddedCount (counter)
- hiveSyncSchemaEvolvedCount (counter)
* Record sync stats in doSync() so GlobalHiveSyncTool reports them too
GlobalHiveSyncTool overrides syncHoodieTable() with a bare doSync(), so it
skipped the stats reset and the total timer. Also explain why
recordPartitionsAdded keeps the max rather than the sum.
* Explain why the sync stats are created in two places
* Initialize the sync duration histogram like the other histograms
* Rename the metastore time to the remaining time
It covers the local work between the metastore calls too, such as
comparing the partitions and schemas found, so hiveSyncMetastoreDurationMs
and getMetastoreMs() become hiveSyncRemainingDurationMs and
getRemainingMs().
* Count the partitions added batch by batch
The partitions were counted only after every batch was added, so when a
later batch failed, the batches already added were never counted: the
retry does not add them again. HoodieMetaSyncOperations and DDLExecutor
get an addPartitionsToTable overload that reports each batch as the
metastore takes it, which the HMS, HiveQL and JDBC executors implement,
including on the HiveQL driver pool.
* Record the init time when building the sync tool fails
Building it fails when the metastore cannot be reached, which is when
the time spent there matters most.
* Return the sync tool after recording its init time
* Drop the overloads that only passed a no-op callback
runSQLs, HiveDriverPool.dispatchAll and DDLExecutor.addPartitionsToTable
each keep one method taking the callback, and callers that do not count
pass a no-op. HoodieMetaSyncOperations keeps both addPartitionsToTable
methods, since sync clients outside hudi-hive-sync implement the one
without a callback. Also rename the callback to onStatementSucceeded,
since it is only called for statements that succeed.
* Say that the last batch of partitions can be smaller
* Add an option to enable the Hive sync stats, off by default
* perf(flink): skip disabled Hive sync histograms and rename stats
---------
Co-authored-by: danny0405 <[email protected]>
---
.../apache/hudi/metrics/FlinkHiveSyncMetrics.java | 101 +++++++++++++-
.../hudi/sink/StreamWriteOperatorCoordinator.java | 26 +++-
.../apache/hudi/sink/utils/HiveSyncContext.java | 10 ++
.../hudi/metrics/TestFlinkHiveSyncMetrics.java | 97 ++++++++++++-
.../sink/TestStreamWriteOperatorCoordinator.java | 34 ++++-
.../org/apache/hudi/hive/HiveSyncConfigHolder.java | 11 ++
.../java/org/apache/hudi/hive/HiveSyncTool.java | 40 +++++-
.../org/apache/hudi/hive/HoodieHiveSyncStats.java | 152 +++++++++++++++++++++
.../org/apache/hudi/hive/TestHiveSyncTool.java | 138 +++++++++++++++++++
.../apache/hudi/hive/TestHoodieHiveSyncStats.java | 128 +++++++++++++++++
10 files changed, 723 insertions(+), 14 deletions(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/metrics/FlinkHiveSyncMetrics.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/metrics/FlinkHiveSyncMetrics.java
index cc25e6080dce..dd1a6d99127d 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/metrics/FlinkHiveSyncMetrics.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/metrics/FlinkHiveSyncMetrics.java
@@ -18,6 +18,8 @@
package org.apache.hudi.metrics;
+import org.apache.hudi.hive.HoodieHiveSyncStats;
+
import com.codahale.metrics.SlidingWindowReservoir;
import org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper;
import org.apache.flink.metrics.Counter;
@@ -32,6 +34,9 @@ import org.apache.flink.metrics.ThreadSafeSimpleCounter;
* <p>A sync runs on the coordinator's Hive sync executor after a checkpoint
commit, and on its event
* executor after the commit that ends a bounded input, so each metric may be
updated from either
* thread.
+ *
+ * <p>The breakdown of each sync, from the init duration to the schema evolved
count, is only
+ * registered when {@code
hoodie.datasource.hive_sync.detailed_metrics.enabled} is set.
*/
public class FlinkHiveSyncMetrics extends HoodieFlinkMetrics {
private static final int HISTOGRAM_WINDOW_SIZE = 100;
@@ -40,6 +45,12 @@ public class FlinkHiveSyncMetrics extends HoodieFlinkMetrics
{
static final String HIVE_SYNC_FAILURE_COUNT = "hiveSyncFailureCount";
static final String HIVE_SYNC_DURATION_MS = "hiveSyncDurationMs";
static final String HIVE_SYNC_LAST_SUCCESS_TIME_MS =
"hiveSyncLastSuccessTimeMs";
+ static final String HIVE_SYNC_INIT_DURATION_MS = "hiveSyncInitDurationMs";
+ static final String HIVE_SYNC_SCHEMA_READ_DURATION_MS =
"hiveSyncSchemaReadDurationMs";
+ static final String HIVE_SYNC_PARTITION_SCAN_DURATION_MS =
"hiveSyncPartitionScanDurationMs";
+ static final String HIVE_SYNC_REMAINING_DURATION_MS =
"hiveSyncRemainingDurationMs";
+ static final String HIVE_SYNC_PARTITIONS_ADDED_COUNT =
"hiveSyncPartitionsAddedCount";
+ static final String HIVE_SYNC_SCHEMA_EVOLVED_COUNT =
"hiveSyncSchemaEvolvedCount";
/** Number of syncs that completed. */
private final Counter syncSuccessCount = new ThreadSafeSimpleCounter();
@@ -56,15 +67,61 @@ public class FlinkHiveSyncMetrics extends
HoodieFlinkMetrics {
* Duration of each sync in milliseconds, failed ones included: a sync that
fails after retrying
* an unreachable metastore is the slow case this is meant to show.
*/
- private final Histogram syncDurationMs;
+ private final Histogram syncDurationMs = newHistogram();
/** Wall-clock time of the last completed sync in epoch milliseconds, or 0
until the first one. */
private volatile long lastSyncSuccessTimeMs = 0L;
- public FlinkHiveSyncMetrics(MetricGroup metricGroup) {
+ /**
+ * Time to build the sync tool, which creates the Hudi meta client and
connects to the metastore.
+ * Recorded whether building it succeeds or fails, so the time spent failing
to reach the
+ * metastore shows here.
+ */
+ private final Histogram initDurationMs;
+
+ /** Time a sync spent reading the table schema from storage, for syncs that
read it. */
+ private final Histogram schemaReadDurationMs;
+
+ /**
+ * Time a sync spent finding the partitions to sync, from the timeline or by
listing storage, for
+ * syncs that looked for them. This grows with the number of commits a sync
has to catch up on.
+ */
+ private final Histogram partitionScanDurationMs;
+
+ /**
+ * Time a sync spent on everything but reading the schema and finding
partitions: mostly metastore
+ * calls, plus the local work between them, such as comparing the partitions
and schemas found.
+ */
+ private final Histogram remainingDurationMs;
+
+ /**
+ * Number of partitions added to the metastore, counting for each sync the
most added to any one of
+ * its Hive tables. A MERGE_ON_READ table normally adds the same partitions
to each of its Hive
+ * tables, and they are counted once. It staying flat while commits land
means the syncs register
+ * nothing.
+ */
+ private final Counter partitionsAddedCount = new ThreadSafeSimpleCounter();
+
+ /**
+ * Number of syncs that found the storage schema changed and pushed it to
the metastore, counted
+ * once per sync however many Hive tables the table is registered as.
+ */
+ private final Counter schemaEvolvedCount = new ThreadSafeSimpleCounter();
+
+ private final boolean detailedMetricsEnabled;
+
+ /**
+ * @param detailedMetricsEnabled whether to register and update the
breakdown of each sync, which the sync records
+ * only when its detailed metrics are enabled
+ */
+ public FlinkHiveSyncMetrics(MetricGroup metricGroup, boolean
detailedMetricsEnabled) {
super(metricGroup);
- this.syncDurationMs = new DropwizardHistogramWrapper(
- new com.codahale.metrics.Histogram(new
SlidingWindowReservoir(HISTOGRAM_WINDOW_SIZE)));
+ this.detailedMetricsEnabled = detailedMetricsEnabled;
+ // Detailed histograms are only accessed when detailed metrics are enabled.
+ this.initDurationMs = detailedMetricsEnabled ? newHistogram() : null;
+ this.schemaReadDurationMs = detailedMetricsEnabled ? newHistogram() : null;
+ this.partitionScanDurationMs = detailedMetricsEnabled ? newHistogram() :
null;
+ this.remainingDurationMs = detailedMetricsEnabled ? newHistogram() : null;
}
@Override
@@ -73,6 +130,37 @@ public class FlinkHiveSyncMetrics extends
HoodieFlinkMetrics {
metricGroup.counter(HIVE_SYNC_FAILURE_COUNT, syncFailureCount);
metricGroup.histogram(HIVE_SYNC_DURATION_MS, syncDurationMs);
metricGroup.gauge(HIVE_SYNC_LAST_SUCCESS_TIME_MS, () ->
lastSyncSuccessTimeMs);
+ if (detailedMetricsEnabled) {
+ metricGroup.histogram(HIVE_SYNC_INIT_DURATION_MS, initDurationMs);
+ metricGroup.histogram(HIVE_SYNC_SCHEMA_READ_DURATION_MS,
schemaReadDurationMs);
+ metricGroup.histogram(HIVE_SYNC_PARTITION_SCAN_DURATION_MS,
partitionScanDurationMs);
+ metricGroup.histogram(HIVE_SYNC_REMAINING_DURATION_MS,
remainingDurationMs);
+ metricGroup.counter(HIVE_SYNC_PARTITIONS_ADDED_COUNT,
partitionsAddedCount);
+ metricGroup.counter(HIVE_SYNC_SCHEMA_EVOLVED_COUNT, schemaEvolvedCount);
+ }
+ }
+
+ public void updateInitDuration(long durationMs) {
+ if (!detailedMetricsEnabled) {
+ return;
+ }
+ initDurationMs.update(durationMs);
+ }
+
+ /**
+ * Records what a sync did, after it completed or failed.
+ */
+ public void updateSyncMetrics(HoodieHiveSyncStats syncMetrics) {
+ if (!detailedMetricsEnabled) {
+ return;
+ }
+ syncMetrics.getSchemaReadMs().ifPresent(schemaReadDurationMs::update);
+
syncMetrics.getPartitionScanMs().ifPresent(partitionScanDurationMs::update);
+ syncMetrics.getRemainingMs().ifPresent(remainingDurationMs::update);
+ partitionsAddedCount.inc(syncMetrics.getPartitionsAdded());
+ if (syncMetrics.isSchemaEvolved()) {
+ schemaEvolvedCount.inc();
+ }
}
public void markSyncSucceeded(long durationMs) {
@@ -85,4 +173,9 @@ public class FlinkHiveSyncMetrics extends HoodieFlinkMetrics
{
syncDurationMs.update(durationMs);
syncFailureCount.inc();
}
+
+ private static Histogram newHistogram() {
+ return new DropwizardHistogramWrapper(
+ new com.codahale.metrics.Histogram(new
SlidingWindowReservoir(HISTOGRAM_WINDOW_SIZE)));
+ }
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java
index 5a56001e6437..595578c05d7f 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java
@@ -475,7 +475,7 @@ public class StreamWriteOperatorCoordinator
.threadFactory(getThreadFactory("hive-sync"))
.waitForTasksFinish(true).build();
this.hiveSyncContext = HiveSyncContext.create(conf, this.storageConf);
- this.hiveSyncMetrics = new FlinkHiveSyncMetrics(context.metricGroup());
+ this.hiveSyncMetrics = new FlinkHiveSyncMetrics(context.metricGroup(),
hiveSyncContext.isDetailedMetricsEnabled());
this.hiveSyncMetrics.registerMetrics();
}
@@ -497,12 +497,16 @@ public class StreamWriteOperatorCoordinator
*/
public void doSyncHive() {
HoodieTimer timer = HoodieTimer.start();
- try (HiveSyncTool syncTool = hiveSyncContext.hiveSyncTool()) {
+ try (HiveSyncTool syncTool = createHiveSyncTool()) {
if (!syncTool.isSyncClientInitialized()) {
hiveSyncMetrics.markSyncFailed(timer.endTimer());
return;
}
- syncTool.syncHoodieTable();
+ try {
+ syncTool.syncHoodieTable();
+ } finally {
+ hiveSyncMetrics.updateSyncMetrics(syncTool.getSyncMetrics());
+ }
} catch (Throwable t) {
hiveSyncMetrics.markSyncFailed(timer.endTimer());
throw t;
@@ -510,6 +514,22 @@ public class StreamWriteOperatorCoordinator
hiveSyncMetrics.markSyncSucceeded(timer.endTimer());
}
+ /**
+ * Builds the sync tool. With detailed metrics enabled, records how long
that took even when it
+ * fails, as when the metastore cannot be reached.
+ */
+ private HiveSyncTool createHiveSyncTool() {
+ if (!hiveSyncContext.isDetailedMetricsEnabled()) {
+ return hiveSyncContext.hiveSyncTool();
+ }
+ HoodieTimer timer = HoodieTimer.start();
+ try {
+ return hiveSyncContext.hiveSyncTool();
+ } finally {
+ hiveSyncMetrics.updateInitDuration(timer.endTimer());
+ }
+ }
+
private void scheduleTableServices(Boolean committed) {
// if compaction is on, schedule the compaction
if (tableState.scheduleCompaction) {
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/HiveSyncContext.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/HiveSyncContext.java
index 349084e128d9..cc626b5e342c 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/HiveSyncContext.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/HiveSyncContext.java
@@ -20,6 +20,7 @@ package org.apache.hudi.sink.utils;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.ConfigUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.configuration.FlinkOptions;
@@ -42,6 +43,7 @@ import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_IGNORE_EXCEPTIONS;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_PASS;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SKIP_RO_SUFFIX_FOR_READ_OPTIMIZED_TABLE;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SUPPORT_TIMESTAMP_TYPE;
+import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_DETAILED_METRICS_ENABLED;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_MODE;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_TABLE_STRATEGY;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_TABLE_PROPERTIES;
@@ -77,6 +79,14 @@ public class HiveSyncContext {
this.hiveConf = hiveConf;
}
+ /**
+ * Whether the sync tools of this context record detailed metrics. Read from
the same properties as
+ * the sync tools, which accept the option with or without the {@code
properties.} prefix.
+ */
+ public boolean isDetailedMetricsEnabled() {
+ return ConfigUtils.getBooleanWithAltKeys(props,
HIVE_SYNC_DETAILED_METRICS_ENABLED);
+ }
+
public HiveSyncTool hiveSyncTool() {
HiveSyncMode syncMode =
HiveSyncMode.of(props.getProperty(HIVE_SYNC_MODE.key()));
if (syncMode == HiveSyncMode.GLUE) {
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/metrics/TestFlinkHiveSyncMetrics.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/metrics/TestFlinkHiveSyncMetrics.java
index 536999adb79e..9026ea0d2219 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/metrics/TestFlinkHiveSyncMetrics.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/metrics/TestFlinkHiveSyncMetrics.java
@@ -18,6 +18,9 @@
package org.apache.hudi.metrics;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.hive.HoodieHiveSyncStats;
+
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.Histogram;
@@ -35,11 +38,20 @@ import java.util.concurrent.TimeUnit;
import static
org.apache.hudi.metrics.FlinkHiveSyncMetrics.HIVE_SYNC_DURATION_MS;
import static
org.apache.hudi.metrics.FlinkHiveSyncMetrics.HIVE_SYNC_FAILURE_COUNT;
+import static
org.apache.hudi.metrics.FlinkHiveSyncMetrics.HIVE_SYNC_INIT_DURATION_MS;
import static
org.apache.hudi.metrics.FlinkHiveSyncMetrics.HIVE_SYNC_LAST_SUCCESS_TIME_MS;
+import static
org.apache.hudi.metrics.FlinkHiveSyncMetrics.HIVE_SYNC_PARTITIONS_ADDED_COUNT;
+import static
org.apache.hudi.metrics.FlinkHiveSyncMetrics.HIVE_SYNC_PARTITION_SCAN_DURATION_MS;
+import static
org.apache.hudi.metrics.FlinkHiveSyncMetrics.HIVE_SYNC_REMAINING_DURATION_MS;
+import static
org.apache.hudi.metrics.FlinkHiveSyncMetrics.HIVE_SYNC_SCHEMA_EVOLVED_COUNT;
+import static
org.apache.hudi.metrics.FlinkHiveSyncMetrics.HIVE_SYNC_SCHEMA_READ_DURATION_MS;
import static
org.apache.hudi.metrics.FlinkHiveSyncMetrics.HIVE_SYNC_SUCCESS_COUNT;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/**
* Tests for {@link FlinkHiveSyncMetrics}.
@@ -52,7 +64,7 @@ class TestFlinkHiveSyncMetrics {
@BeforeEach
void setUp() {
metricGroup = new CapturingMetricGroup();
- metrics = new FlinkHiveSyncMetrics(metricGroup);
+ metrics = new FlinkHiveSyncMetrics(metricGroup, true);
metrics.registerMetrics();
}
@@ -62,6 +74,29 @@ class TestFlinkHiveSyncMetrics {
assertNotNull(metricGroup.counters.get(HIVE_SYNC_FAILURE_COUNT));
assertNotNull(metricGroup.histograms.get(HIVE_SYNC_DURATION_MS));
assertNotNull(metricGroup.gauges.get(HIVE_SYNC_LAST_SUCCESS_TIME_MS));
+ assertNotNull(metricGroup.histograms.get(HIVE_SYNC_INIT_DURATION_MS));
+
assertNotNull(metricGroup.histograms.get(HIVE_SYNC_SCHEMA_READ_DURATION_MS));
+
assertNotNull(metricGroup.histograms.get(HIVE_SYNC_PARTITION_SCAN_DURATION_MS));
+ assertNotNull(metricGroup.histograms.get(HIVE_SYNC_REMAINING_DURATION_MS));
+ assertNotNull(metricGroup.counters.get(HIVE_SYNC_PARTITIONS_ADDED_COUNT));
+ assertNotNull(metricGroup.counters.get(HIVE_SYNC_SCHEMA_EVOLVED_COUNT));
+ }
+
+ @Test
+ void testBreakdownMetricsNotRegisteredWhenDetailedMetricsDisabled() {
+ CapturingMetricGroup group = new CapturingMetricGroup();
+ new FlinkHiveSyncMetrics(group, false).registerMetrics();
+
+ assertNotNull(group.counters.get(HIVE_SYNC_SUCCESS_COUNT));
+ assertNotNull(group.counters.get(HIVE_SYNC_FAILURE_COUNT));
+ assertNotNull(group.histograms.get(HIVE_SYNC_DURATION_MS));
+ assertNotNull(group.gauges.get(HIVE_SYNC_LAST_SUCCESS_TIME_MS));
+ assertNull(group.histograms.get(HIVE_SYNC_INIT_DURATION_MS));
+ assertNull(group.histograms.get(HIVE_SYNC_SCHEMA_READ_DURATION_MS));
+ assertNull(group.histograms.get(HIVE_SYNC_PARTITION_SCAN_DURATION_MS));
+ assertNull(group.histograms.get(HIVE_SYNC_REMAINING_DURATION_MS));
+ assertNull(group.counters.get(HIVE_SYNC_PARTITIONS_ADDED_COUNT));
+ assertNull(group.counters.get(HIVE_SYNC_SCHEMA_EVOLVED_COUNT));
}
@Test
@@ -70,6 +105,47 @@ class TestFlinkHiveSyncMetrics {
assertEquals(0, failureCount());
assertEquals(0, duration().getCount());
assertEquals(0L, lastSuccessTime());
+ assertEquals(0, histogram(HIVE_SYNC_INIT_DURATION_MS).getCount());
+ assertEquals(0, histogram(HIVE_SYNC_SCHEMA_READ_DURATION_MS).getCount());
+ assertEquals(0,
histogram(HIVE_SYNC_PARTITION_SCAN_DURATION_MS).getCount());
+ assertEquals(0, histogram(HIVE_SYNC_REMAINING_DURATION_MS).getCount());
+ assertEquals(0, counter(HIVE_SYNC_PARTITIONS_ADDED_COUNT));
+ assertEquals(0, counter(HIVE_SYNC_SCHEMA_EVOLVED_COUNT));
+ }
+
+ @Test
+ void testUpdateInitDuration() {
+ metrics.updateInitDuration(7L);
+
+ assertEquals(1, histogram(HIVE_SYNC_INIT_DURATION_MS).getCount());
+ assertEquals(7L,
histogram(HIVE_SYNC_INIT_DURATION_MS).getStatistics().getMax());
+ }
+
+ @Test
+ void testUpdateSyncMetrics() {
+ metrics.updateSyncMetrics(stats(Option.of(10L), Option.of(20L),
Option.of(30L), 4, true));
+ metrics.updateSyncMetrics(stats(Option.of(11L), Option.of(21L),
Option.of(31L), 2, false));
+
+ assertEquals(2, histogram(HIVE_SYNC_SCHEMA_READ_DURATION_MS).getCount());
+ assertEquals(11L,
histogram(HIVE_SYNC_SCHEMA_READ_DURATION_MS).getStatistics().getMax());
+ assertEquals(2,
histogram(HIVE_SYNC_PARTITION_SCAN_DURATION_MS).getCount());
+ assertEquals(21L,
histogram(HIVE_SYNC_PARTITION_SCAN_DURATION_MS).getStatistics().getMax());
+ assertEquals(2, histogram(HIVE_SYNC_REMAINING_DURATION_MS).getCount());
+ assertEquals(31L,
histogram(HIVE_SYNC_REMAINING_DURATION_MS).getStatistics().getMax());
+ assertEquals(6, counter(HIVE_SYNC_PARTITIONS_ADDED_COUNT));
+ assertEquals(1, counter(HIVE_SYNC_SCHEMA_EVOLVED_COUNT));
+ }
+
+ @Test
+ void testUpdateSyncMetricsSkipsStepsThatDidNotRun() {
+ metrics.updateSyncMetrics(stats(Option.empty(), Option.empty(),
Option.of(5L), 0, false));
+
+ assertEquals(0, histogram(HIVE_SYNC_SCHEMA_READ_DURATION_MS).getCount(),
+ "A sync that read no schema must not record a schema read of 0 ms");
+ assertEquals(0,
histogram(HIVE_SYNC_PARTITION_SCAN_DURATION_MS).getCount());
+ assertEquals(1, histogram(HIVE_SYNC_REMAINING_DURATION_MS).getCount());
+ assertEquals(0, counter(HIVE_SYNC_PARTITIONS_ADDED_COUNT));
+ assertEquals(0, counter(HIVE_SYNC_SCHEMA_EVOLVED_COUNT));
}
@Test
@@ -175,6 +251,25 @@ class TestFlinkHiveSyncMetrics {
return (Long)
metricGroup.gauges.get(HIVE_SYNC_LAST_SUCCESS_TIME_MS).getValue();
}
+ private Histogram histogram(String name) {
+ return metricGroup.histograms.get(name);
+ }
+
+ private long counter(String name) {
+ return metricGroup.counters.get(name).getCount();
+ }
+
+ private static HoodieHiveSyncStats stats(Option<Long> schemaReadMs,
Option<Long> partitionScanMs,
+ Option<Long> remainingMs, int
partitionsAdded, boolean schemaEvolved) {
+ HoodieHiveSyncStats stats = mock(HoodieHiveSyncStats.class);
+ when(stats.getSchemaReadMs()).thenReturn(schemaReadMs);
+ when(stats.getPartitionScanMs()).thenReturn(partitionScanMs);
+ when(stats.getRemainingMs()).thenReturn(remainingMs);
+ when(stats.getPartitionsAdded()).thenReturn(partitionsAdded);
+ when(stats.isSchemaEvolved()).thenReturn(schemaEvolved);
+ return stats;
+ }
+
private static class CapturingMetricGroup extends UnregisteredMetricsGroup {
private final Map<String, Counter> counters = new HashMap<>();
private final Map<String, Histogram> histograms = new HashMap<>();
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
index b32b62d6dfbc..5664dcbb9c60 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
@@ -48,6 +48,7 @@ import org.apache.hudi.configuration.HadoopConfigurations;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.MissingSchemaFieldException;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
+import org.apache.hudi.hive.HiveSyncConfigHolder;
import org.apache.hudi.hive.HoodieHiveSyncException;
import org.apache.hudi.metadata.HoodieTableMetadata;
import org.apache.hudi.metadata.MetadataPartitionType;
@@ -711,9 +712,32 @@ public class TestStreamWriteOperatorCoordinator {
}
@Test
- void testHiveSyncFailureUpdatesMetrics() throws Exception {
+ void testHiveSyncBreakdownMetricsNotRegisteredUnlessDetailedMetricsEnabled()
throws Exception {
+ coordinator.close();
+ CapturingCoordinatorMetricGroup metricGroup = new
CapturingCoordinatorMetricGroup();
+ coordinator = startCoordinator(getUnreachableMetastoreConf(), metricGroup);
+
+ assertThrows(HoodieHiveSyncException.class, () ->
coordinator.doSyncHive());
+
+ assertEquals(1,
metricGroup.counters.get("hiveSyncFailureCount").getCount(),
+ "A failed sync is counted whether or not the detailed metrics are
enabled");
+ assertNotNull(metricGroup.counters.get("hiveSyncSuccessCount"));
+ assertNotNull(metricGroup.histograms.get("hiveSyncDurationMs"));
+ assertNotNull(metricGroup.gauges.get("hiveSyncLastSuccessTimeMs"));
+ assertNull(metricGroup.histograms.get("hiveSyncInitDurationMs"));
+ assertNull(metricGroup.histograms.get("hiveSyncSchemaReadDurationMs"));
+ assertNull(metricGroup.histograms.get("hiveSyncPartitionScanDurationMs"));
+ assertNull(metricGroup.histograms.get("hiveSyncRemainingDurationMs"));
+ assertNull(metricGroup.counters.get("hiveSyncPartitionsAddedCount"));
+ assertNull(metricGroup.counters.get("hiveSyncSchemaEvolvedCount"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"", "properties."})
+ void testHiveSyncFailureUpdatesMetrics(String keyPrefix) throws Exception {
coordinator.close();
Configuration conf = getUnreachableMetastoreConf();
+ conf.setString(keyPrefix +
HiveSyncConfigHolder.HIVE_SYNC_DETAILED_METRICS_ENABLED.key(), "true");
CapturingCoordinatorMetricGroup metricGroup = new
CapturingCoordinatorMetricGroup();
coordinator = startCoordinator(conf, metricGroup);
@@ -726,6 +750,9 @@ public class TestStreamWriteOperatorCoordinator {
assertEquals(2,
metricGroup.counters.get("hiveSyncFailureCount").getCount());
assertEquals(2,
metricGroup.histograms.get("hiveSyncDurationMs").getCount());
assertEquals(0L,
metricGroup.gauges.get("hiveSyncLastSuccessTimeMs").getValue());
+ assertEquals(2,
metricGroup.histograms.get("hiveSyncInitDurationMs").getCount(),
+ "The time spent failing to build the sync tool is init time");
+ assertEquals(0,
metricGroup.histograms.get("hiveSyncRemainingDurationMs").getCount());
}
@Test
@@ -733,6 +760,7 @@ public class TestStreamWriteOperatorCoordinator {
coordinator.close();
Configuration conf = getUnreachableMetastoreConf();
conf.set(FlinkOptions.HIVE_SYNC_IGNORE_EXCEPTIONS, true);
+
conf.setString(HiveSyncConfigHolder.HIVE_SYNC_DETAILED_METRICS_ENABLED.key(),
"true");
CapturingCoordinatorMetricGroup metricGroup = new
CapturingCoordinatorMetricGroup();
coordinator = startCoordinator(conf, metricGroup);
@@ -742,6 +770,10 @@ public class TestStreamWriteOperatorCoordinator {
assertEquals(1,
metricGroup.counters.get("hiveSyncFailureCount").getCount());
assertEquals(1,
metricGroup.histograms.get("hiveSyncDurationMs").getCount());
assertEquals(0L,
metricGroup.gauges.get("hiveSyncLastSuccessTimeMs").getValue());
+ assertEquals(1,
metricGroup.histograms.get("hiveSyncInitDurationMs").getCount(),
+ "The time spent failing to connect is init time");
+ assertEquals(0,
metricGroup.histograms.get("hiveSyncRemainingDurationMs").getCount(),
+ "A sync that never ran records no remaining time");
}
private Configuration getUnreachableMetastoreConf() {
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfigHolder.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfigHolder.java
index 1c1ed1895ed6..a2b7d4c07757 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfigHolder.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfigHolder.java
@@ -150,6 +150,17 @@ public class HiveSyncConfigHolder {
+ "`hoodie.datasource.hive_sync.batching.enabled` is true. The same
value sizes both pools: the "
+ "Hive Driver workers used for ADD/TOUCH/SET_LOCATION and the
metastore (Thrift) clients used "
+ "for DROP. Ignored otherwise.");
+ public static final ConfigProperty<Boolean>
HIVE_SYNC_DETAILED_METRICS_ENABLED = ConfigProperty
+ .key("hoodie.datasource.hive_sync.detailed_metrics.enabled")
+ .defaultValue(false)
+ .markAdvanced()
+ .sinceVersion("1.3.0")
+ .withDocumentation("When true, each Hive sync records a breakdown of
what it did: the time spent "
+ + "reading the table schema, the time spent finding the partitions
to sync, the time spent on "
+ + "the rest, the most partitions added to any one Hive table, and
whether the schema was "
+ + "pushed to the metastore. When false, the sync steps are not
timed. The Flink write "
+ + "coordinator reports the breakdown, and the time to build the sync
tool, as metrics only "
+ + "when this is true.");
public static final ConfigProperty<String> HIVE_SYNC_MODE = ConfigProperty
.key("hoodie.datasource.hive_sync.mode")
.noDefaultValue()
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
index f0e3ee7da2a9..c45f3d1e79c9 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
@@ -66,6 +66,7 @@ import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SKIP_RO_SUFFIX_FOR_
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SUPPORT_TIMESTAMP_TYPE;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_AS_DATA_SOURCE_TABLE;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_COMMENT;
+import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_DETAILED_METRICS_ENABLED;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_OMIT_METADATA_FIELDS;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_SCHEMA_STRING_LENGTH_THRESHOLD;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_TABLE_STRATEGY;
@@ -110,6 +111,11 @@ public class HiveSyncTool extends HoodieSyncTool
implements AutoCloseable {
private String hiveSyncTableStrategy;
+ private final boolean detailedMetricsEnabled;
+
+ // never null, so getSyncMetrics() also works when no sync ran, e.g. when
the metastore client could not be created
+ private HoodieHiveSyncStats syncMetrics;
+
public HiveSyncTool(Properties props, Configuration hadoopConf) {
this(props, hadoopConf, Option.empty());
}
@@ -129,6 +135,8 @@ public class HiveSyncTool extends HoodieSyncTool implements
AutoCloseable {
}
this.config = new HiveSyncConfig(props, hadoopConfForSync);
+ this.detailedMetricsEnabled =
config.getBooleanOrDefault(HIVE_SYNC_DETAILED_METRICS_ENABLED);
+ this.syncMetrics = new HoodieHiveSyncStats(detailedMetricsEnabled);
HoodieTableMetaClient metaClient = metaClientOption.orElseGet(() ->
buildMetaClient(config));
initSyncClient(config, metaClient);
// initSyncClient leaves syncClient as null
@@ -198,7 +206,20 @@ public class HiveSyncTool extends HoodieSyncTool
implements AutoCloseable {
return syncClient != null;
}
+ /**
+ * Returns what the last {@link #syncHoodieTable()} did, including when it
failed.
+ */
+ public HoodieHiveSyncStats getSyncMetrics() {
+ return syncMetrics;
+ }
+
protected void doSync() {
+ // the durations add up across the Hive tables of one sync, so start fresh
rather than adding to the previous sync's
+ syncMetrics = new HoodieHiveSyncStats(detailedMetricsEnabled);
+ syncMetrics.timeTotal(this::syncTables);
+ }
+
+ private void syncTables() {
// create database if needed
checkAndCreateDatabase();
@@ -271,7 +292,7 @@ public class HiveSyncTool extends HoodieSyncTool implements
AutoCloseable {
return;
}
// Get the parquet schema for this table looking at the latest commit
- HoodieSchema schema =
syncClient.getStorageSchema(!config.getBoolean(HIVE_SYNC_OMIT_METADATA_FIELDS));
+ HoodieSchema schema = readStorageSchema();
boolean schemaChanged;
boolean propertiesChanged;
@@ -325,6 +346,11 @@ public class HiveSyncTool extends HoodieSyncTool
implements AutoCloseable {
return compareTimestamps(lastCommitTimeSynced.get(), LESSER_THAN,
midpointInstantTime);
}
+ private HoodieSchema readStorageSchema() {
+ return syncMetrics.timeSchemaRead(
+ () ->
syncClient.getStorageSchema(!config.getBoolean(HIVE_SYNC_OMIT_METADATA_FIELDS)));
+ }
+
private boolean isAlreadySynced(String tableName) {
return syncClient.getLastCommitTimeSynced(tableName)
.map(lastCommit -> {
@@ -382,12 +408,14 @@ public class HiveSyncTool extends HoodieSyncTool
implements AutoCloseable {
config.getString(META_SYNC_BASE_PATH), config.getHadoopFileSystem());
partitionsChanged = syncAllPartitions(tableName);
} else {
- List<String> writtenPartitionsSince =
syncClient.getWrittenPartitionsSince(lastCommitTimeSynced,
lastCommitCompletionTimeSynced);
+ List<String> writtenPartitionsSince = syncMetrics.timePartitionScan(
+ () -> syncClient.getWrittenPartitionsSince(lastCommitTimeSynced,
lastCommitCompletionTimeSynced));
log.info("Storage partitions scan complete. Found {}",
writtenPartitionsSince.size());
// Sync the partitions if needed
// find dropped partitions, if any, in the latest commit
- Set<String> droppedPartitions =
syncClient.getDroppedPartitionsSince(lastCommitTimeSynced,
lastCommitCompletionTimeSynced);
+ Set<String> droppedPartitions = syncMetrics.timePartitionScan(
+ () -> syncClient.getDroppedPartitionsSince(lastCommitTimeSynced,
lastCommitCompletionTimeSynced));
log.info("Partitions dropped since last sync: {}",
droppedPartitions.size());
partitionsChanged = syncPartitions(tableName, writtenPartitionsSince,
droppedPartitions);
}
@@ -422,7 +450,7 @@ public class HiveSyncTool extends HoodieSyncTool implements
AutoCloseable {
private void recreateAndSyncHiveTable(String tableName, boolean
useRealtimeInputFormat, boolean readAsOptimized) {
log.info("recreating and syncing the table {}", tableName);
Timer.Context timerContext = metrics.getRecreateAndSyncTimer();
- HoodieSchema schema =
syncClient.getStorageSchema(!config.getBoolean(HIVE_SYNC_OMIT_METADATA_FIELDS));
+ HoodieSchema schema = readStorageSchema();
try {
createOrReplaceTable(tableName, useRealtimeInputFormat, readAsOptimized,
schema);
syncAllPartitions(tableName);
@@ -498,6 +526,7 @@ public class HiveSyncTool extends HoodieSyncTool implements
AutoCloseable {
} else {
log.info("Schema difference found for {}. Updated schema: {}",
tableName, schema);
syncClient.updateTableSchema(tableName, schema, schemaDiff);
+ syncMetrics.markSchemaEvolved();
schemaChanged = true;
}
@@ -552,7 +581,7 @@ public class HiveSyncTool extends HoodieSyncTool implements
AutoCloseable {
}
List<Partition> allPartitionsInMetastore =
syncClient.getAllPartitions(tableName);
- List<String> allPartitionsOnStorage =
syncClient.getAllPartitionPathsOnStorage();
+ List<String> allPartitionsOnStorage =
syncMetrics.timePartitionScan(syncClient::getAllPartitionPathsOnStorage);
return syncPartitions(
tableName,
syncClient.getPartitionEvents(allPartitionsInMetastore,
allPartitionsOnStorage));
@@ -600,6 +629,7 @@ public class HiveSyncTool extends HoodieSyncTool implements
AutoCloseable {
if (!newPartitions.isEmpty()) {
log.info("New Partitions {}", newPartitions);
syncClient.addPartitionsToTable(tableName, newPartitions);
+ syncMetrics.recordPartitionsAdded(newPartitions.size());
}
List<String> updatePartitions = filterPartitions(partitionEventList,
PartitionEventType.UPDATE);
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncStats.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncStats.java
new file mode 100644
index 000000000000..f17d9d49e325
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncStats.java
@@ -0,0 +1,152 @@
+/*
+ * 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.hive;
+
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.Option;
+
+import java.util.function.LongConsumer;
+import java.util.function.Supplier;
+
+/**
+ * What one {@link HiveSyncTool#syncHoodieTable()} did, for callers that
report on their syncs.
+ *
+ * <p>A MERGE_ON_READ table is registered as several Hive tables, all synced
by one call. The
+ * durations add up over those tables. The partition count is the most
partitions added to any one
+ * of them, and the schema flag is set when any of them had its schema evolved.
+ *
+ * <p>After a failed sync, the metrics cover the steps that ran before the
failure.
+ *
+ * <p>Unless {@code hoodie.datasource.hive_sync.detailed_metrics.enabled} is
set, the sync steps run
+ * untimed and nothing is recorded, so the metrics stay empty.
+ */
+public class HoodieHiveSyncStats {
+
+ private final boolean detailedMetricsEnabled;
+ private long totalMs = -1;
+ private long schemaReadMs = -1;
+ private long partitionScanMs = -1;
+ private int partitionsAdded = 0;
+ private boolean schemaEvolved = false;
+
+ /**
+ * @param detailedMetricsEnabled whether to time the sync steps and keep
what is recorded; disabled
+ * metrics run the steps untimed and stay empty
+ */
+ public HoodieHiveSyncStats(boolean detailedMetricsEnabled) {
+ this.detailedMetricsEnabled = detailedMetricsEnabled;
+ }
+
+ /**
+ * Time spent reading the table schema from storage, or empty when no step
read it, as when the
+ * metastore is already up to date with the latest commit.
+ */
+ public Option<Long> getSchemaReadMs() {
+ return schemaReadMs < 0 ? Option.empty() : Option.of(schemaReadMs);
+ }
+
+ /**
+ * Time spent finding the partitions to sync, from the timeline or by
listing storage, or empty
+ * when no step looked for them.
+ */
+ public Option<Long> getPartitionScanMs() {
+ return partitionScanMs < 0 ? Option.empty() : Option.of(partitionScanMs);
+ }
+
+ /**
+ * Time spent on the rest of the sync: mostly the calls to the metastore,
checking for the database
+ * and tables, creating or altering them, and reading and writing their
partitions, but also the
+ * local work between them, such as comparing the partitions and schemas
found. Empty when the sync
+ * did not run, because the metastore client could not be created.
+ */
+ public Option<Long> getRemainingMs() {
+ if (totalMs < 0) {
+ return Option.empty();
+ }
+ return Option.of(Math.max(0, totalMs - Math.max(0, schemaReadMs) -
Math.max(0, partitionScanMs)));
+ }
+
+ /**
+ * The most partitions added to any one Hive table of the sync. They are
counted once adding them
+ * all succeeds, so a sync that fails while adding partitions counts none of
them.
+ *
+ * <p>The Hive tables of a MERGE_ON_READ table normally get the same
partitions, which this counts
+ * once. It is not a count of distinct partitions: when the tables were
missing different
+ * partitions, as after one of them was repaired by hand, it counts the
larger of their additions.
+ */
+ public int getPartitionsAdded() {
+ return partitionsAdded;
+ }
+
+ /** Whether the storage schema differed from the metastore's and was pushed
to it. */
+ public boolean isSchemaEvolved() {
+ return schemaEvolved;
+ }
+
+ void timeTotal(Runnable sync) {
+ time(() -> {
+ sync.run();
+ return null;
+ }, this::setTotalMs);
+ }
+
+ <T> T timeSchemaRead(Supplier<T> step) {
+ return time(step, this::addSchemaReadMs);
+ }
+
+ <T> T timePartitionScan(Supplier<T> step) {
+ return time(step, this::addPartitionScanMs);
+ }
+
+ void recordPartitionsAdded(int count) {
+ if (detailedMetricsEnabled) {
+ partitionsAdded = Math.max(partitionsAdded, count);
+ }
+ }
+
+ void markSchemaEvolved() {
+ if (detailedMetricsEnabled) {
+ schemaEvolved = true;
+ }
+ }
+
+ void setTotalMs(long totalMs) {
+ this.totalMs = totalMs;
+ }
+
+ void addSchemaReadMs(long durationMs) {
+ schemaReadMs = Math.max(0, schemaReadMs) + durationMs;
+ }
+
+ void addPartitionScanMs(long durationMs) {
+ partitionScanMs = Math.max(0, partitionScanMs) + durationMs;
+ }
+
+ private <T> T time(Supplier<T> step, LongConsumer recorder) {
+ if (!detailedMetricsEnabled) {
+ return step.get();
+ }
+ HoodieTimer timer = HoodieTimer.start();
+ try {
+ return step.get();
+ } finally {
+ recorder.accept(timer.endTimer());
+ }
+ }
+}
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
index 86fba5a36969..3d575d243772 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
@@ -48,6 +48,7 @@ import
org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat;
import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils;
import org.apache.hudi.hive.ddl.HMSDDLExecutor;
import org.apache.hudi.hive.ddl.HiveSyncMode;
+import org.apache.hudi.hive.replication.GlobalHiveSyncTool;
import org.apache.hudi.hive.testutils.HiveTestUtil;
import org.apache.hudi.hive.util.HiveDriverPool;
import org.apache.hudi.hive.util.HiveMetaStoreClientPool;
@@ -62,6 +63,7 @@ import
org.apache.hudi.sync.common.model.PartitionEvent.PartitionEventType;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.ql.Driver;
@@ -118,6 +120,7 @@ import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_AS_DATA_SOURCE
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_BATCHING_ENABLED;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_BATCHING_THREADS;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_COMMENT;
+import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_DETAILED_METRICS_ENABLED;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_MODE;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_TABLE_STRATEGY;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_TABLE_PROPERTIES;
@@ -145,6 +148,11 @@ import static
org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.AdditionalAnswers.delegatesTo;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.verify;
@@ -1243,6 +1251,129 @@ public class TestHiveSyncTool {
"The last commit that was synced should be 101");
}
+ @Test
+ void testSyncMetricsAreEmptyUnlessEnabled() throws Exception {
+ HiveTestUtil.createCOWTable("100", 5, true);
+ reInitHiveSyncClient();
+
+ HoodieHiveSyncStats stats = syncAndGetMetrics();
+ assertEquals(5,
hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME).size());
+ assertFalse(stats.getSchemaReadMs().isPresent());
+ assertFalse(stats.getPartitionScanMs().isPresent());
+ assertFalse(stats.getRemainingMs().isPresent());
+ assertEquals(0, stats.getPartitionsAdded());
+ assertFalse(stats.isSchemaEvolved());
+ }
+
+ @ParameterizedTest
+ @MethodSource("syncMode")
+ void testSyncMetricsDescribeWhatTheSyncDid(String syncMode) throws Exception
{
+ hiveSyncProps.setProperty(HIVE_SYNC_DETAILED_METRICS_ENABLED.key(),
"true");
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(), syncMode);
+ HiveTestUtil.createCOWTable("100", 5, true);
+ reInitHiveSyncClient();
+
+ HoodieHiveSyncStats stats = syncAndGetMetrics();
+ assertTrue(stats.getSchemaReadMs().isPresent());
+ assertTrue(stats.getPartitionScanMs().isPresent());
+ assertTrue(stats.getRemainingMs().isPresent());
+ assertEquals(5, stats.getPartitionsAdded());
+ assertFalse(stats.isSchemaEvolved(), "Creating the table is not a schema
evolution");
+
+ HiveTestUtil.addCOWPartitions(1, false, true,
ZonedDateTime.now().plusDays(6), "101");
+ reInitHiveSyncClient();
+ stats = syncAndGetMetrics();
+ assertEquals(1, stats.getPartitionsAdded());
+ assertTrue(stats.isSchemaEvolved(), "The partition written with the
evolved schema changes the table's schema");
+
+ stats = syncAndGetMetrics();
+ assertFalse(stats.getSchemaReadMs().isPresent(), "A table already synced
to the latest commit reads no schema");
+ assertFalse(stats.getPartitionScanMs().isPresent(), "A table already
synced to the latest commit looks for no partitions");
+ assertTrue(stats.getRemainingMs().isPresent());
+ assertEquals(0, stats.getPartitionsAdded());
+ assertFalse(stats.isSchemaEvolved());
+ }
+
+ @ParameterizedTest
+ @MethodSource("syncMode")
+ void testSyncMetricsCountAMergeOnReadTableOnce(String syncMode) throws
Exception {
+ hiveSyncProps.setProperty(HIVE_SYNC_DETAILED_METRICS_ENABLED.key(),
"true");
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(), syncMode);
+ HiveTestUtil.createMORTable("100", "101", 5, true, true);
+ reInitHiveSyncClient();
+
+ HoodieHiveSyncStats stats = syncAndGetMetrics();
+ assertTrue(hiveClient.tableExists(HiveTestUtil.TABLE_NAME +
HiveSyncTool.SUFFIX_READ_OPTIMIZED_TABLE));
+ assertTrue(hiveClient.tableExists(HiveTestUtil.TABLE_NAME +
HiveSyncTool.SUFFIX_SNAPSHOT_TABLE));
+ assertEquals(5, stats.getPartitionsAdded(),
+ "The read-optimized and snapshot tables get the same 5 partitions,
which count once");
+ }
+
+ @Test
+ void
testSyncMetricsCountTheMostPartitionsAddedToOneTableOfAMergeOnReadTable()
throws Exception {
+ hiveSyncProps.setProperty(HIVE_SYNC_DETAILED_METRICS_ENABLED.key(),
"true");
+ HiveTestUtil.createMORTable("100", "101", 5, true, true);
+ reInitHiveSyncClient();
+ syncAndGetMetrics();
+ String roTable = HiveTestUtil.TABLE_NAME +
HiveSyncTool.SUFFIX_READ_OPTIMIZED_TABLE;
+ String rtTable = HiveTestUtil.TABLE_NAME +
HiveSyncTool.SUFFIX_SNAPSHOT_TABLE;
+ List<String> partitions = hiveClient.getAllPartitions(roTable).stream()
+ .map(p -> getRelativePartitionPath(new Path(basePath), new
Path(p.getStorageLocation())))
+ .sorted()
+ .collect(Collectors.toList());
+ // the Hive tables miss different partitions, as after one of them was
repaired by hand
+ hiveClient.dropPartitions(roTable, partitions.subList(0, 1));
+ hiveClient.dropPartitions(rtTable, partitions.subList(1, 3));
+ hiveSyncProps.setProperty(META_SYNC_INCREMENTAL.key(), "false");
+ reInitHiveSyncClient();
+
+ HoodieHiveSyncStats stats = syncAndGetMetrics();
+ assertEquals(5, hiveClient.getAllPartitions(roTable).size());
+ assertEquals(5, hiveClient.getAllPartitions(rtTable).size());
+ assertEquals(2, stats.getPartitionsAdded(),
+ "The most partitions added to one table, not the 3 distinct partitions
the sync added");
+ }
+
+ @Test
+ void testSyncMetricsOfGlobalHiveSyncTool() throws Exception {
+ hiveSyncProps.setProperty(HIVE_SYNC_DETAILED_METRICS_ENABLED.key(),
"true");
+ HiveTestUtil.createCOWTable("100", 5, true);
+
+ try (GlobalHiveSyncTool tool = new GlobalHiveSyncTool(hiveSyncProps,
HiveTestUtil.getHiveConf())) {
+ tool.syncHoodieTable();
+ HoodieHiveSyncStats stats = tool.getSyncMetrics();
+ assertTrue(stats.getRemainingMs().isPresent());
+ assertEquals(5, stats.getPartitionsAdded());
+ }
+ }
+
+ @Test
+ void testSyncMetricsCountNoPartitionsWhenAddingThemFails() throws Exception {
+ hiveSyncProps.setProperty(HIVE_SYNC_DETAILED_METRICS_ENABLED.key(),
"true");
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(), HiveSyncMode.HMS.name());
+ hiveSyncProps.setProperty(HIVE_BATCH_SYNC_PARTITION_NUM.key(), "2");
+ HiveTestUtil.createCOWTable("100", 5, true);
+ reInitHiveSyncClient();
+ // the metastore takes the first batch of 2 partitions and fails the second
+ IMetaStoreClient metaStoreClient =
IMetaStoreClientUtil.getMSC(getHiveConf());
+ IMetaStoreClient failingClient = mock(IMetaStoreClient.class,
delegatesTo(metaStoreClient));
+ doAnswer(invocation -> metaStoreClient.add_partitions(
+ invocation.getArgument(0), invocation.getArgument(1),
invocation.getArgument(2)))
+ .doThrow(new MetaException("second batch fails"))
+ .when(failingClient).add_partitions(anyList(), anyBoolean(),
anyBoolean());
+ hiveClient.ddlExecutor = new HMSDDLExecutor(new
HiveSyncConfig(hiveSyncProps, getHiveConf()), failingClient);
+
+ assertThrows(HoodieException.class, () -> hiveSyncTool.syncHoodieTable());
+ assertEquals(0, hiveSyncTool.getSyncMetrics().getPartitionsAdded(),
+ "Partitions are only counted once adding them all succeeds");
+ reInitHiveSyncClient();
+ assertEquals(2,
hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME).size());
+
+ HoodieHiveSyncStats retryMetrics = syncAndGetMetrics();
+ assertEquals(3, retryMetrics.getPartitionsAdded(), "The retry adds only
the partitions the failed sync did not");
+ assertEquals(5,
hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME).size());
+ }
+
@ParameterizedTest
@MethodSource("syncModeAndEnablePushDown")
public void testRecreateCOWTableOnBasePathChange(String syncMode, String
enablePushDown) throws Exception {
@@ -2873,6 +3004,13 @@ public class TestHiveSyncTool {
assertEquals(updatedTable, hiveClient.getTableName(), "Table name in sync
client should match");
}
+ private HoodieHiveSyncStats syncAndGetMetrics() {
+ hiveSyncTool.syncHoodieTable();
+ HoodieHiveSyncStats stats = hiveSyncTool.getSyncMetrics();
+ reInitHiveSyncClient();
+ return stats;
+ }
+
private void reSyncHiveTable() {
hiveSyncTool.syncHoodieTable();
// we need renew the hive client after tool.syncHoodieTable(), because it
will close hive
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHoodieHiveSyncStats.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHoodieHiveSyncStats.java
new file mode 100644
index 000000000000..7b5ba50eab8f
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHoodieHiveSyncStats.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.hive;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for {@link HoodieHiveSyncStats}.
+ */
+class TestHoodieHiveSyncStats {
+
+ @Test
+ void testNothingRecordedBeforeASync() {
+ HoodieHiveSyncStats stats = new HoodieHiveSyncStats(true);
+ assertFalse(stats.getSchemaReadMs().isPresent());
+ assertFalse(stats.getPartitionScanMs().isPresent());
+ assertFalse(stats.getRemainingMs().isPresent());
+ assertEquals(0, stats.getPartitionsAdded());
+ assertFalse(stats.isSchemaEvolved());
+ }
+
+ @Test
+ void testMetastoreTimeIsWhatTheOtherStepsLeave() {
+ HoodieHiveSyncStats stats = new HoodieHiveSyncStats(true);
+ stats.addSchemaReadMs(10);
+ stats.addSchemaReadMs(5);
+ stats.addPartitionScanMs(30);
+ stats.setTotalMs(100);
+
+ assertEquals(15L, stats.getSchemaReadMs().get());
+ assertEquals(30L, stats.getPartitionScanMs().get());
+ assertEquals(55L, stats.getRemainingMs().get());
+ }
+
+ @Test
+ void testMetastoreTimeWithoutOtherSteps() {
+ HoodieHiveSyncStats stats = new HoodieHiveSyncStats(true);
+ stats.setTotalMs(40);
+
+ assertFalse(stats.getSchemaReadMs().isPresent());
+ assertFalse(stats.getPartitionScanMs().isPresent());
+ assertEquals(40L, stats.getRemainingMs().get());
+ }
+
+ @Test
+ void testPartitionsAddedAndSchemaEvolvedDescribeOneTable() {
+ HoodieHiveSyncStats stats = new HoodieHiveSyncStats(true);
+ stats.recordPartitionsAdded(3);
+ stats.recordPartitionsAdded(3);
+ stats.markSchemaEvolved();
+ stats.markSchemaEvolved();
+
+ assertEquals(3, stats.getPartitionsAdded());
+ assertTrue(stats.isSchemaEvolved());
+ }
+
+ @Test
+ void testPartitionsAddedIsTheMostAddedToOneTable() {
+ HoodieHiveSyncStats stats = new HoodieHiveSyncStats(true);
+ stats.recordPartitionsAdded(2);
+ stats.recordPartitionsAdded(3);
+ stats.recordPartitionsAdded(1);
+
+ assertEquals(3, stats.getPartitionsAdded());
+ }
+
+ @Test
+ void testTimedStepsRecordTheirDurations() {
+ HoodieHiveSyncStats stats = new HoodieHiveSyncStats(true);
+ stats.timeTotal(() -> {
+ assertEquals("schema", stats.timeSchemaRead(() -> "schema"));
+ assertEquals("partitions", stats.timePartitionScan(() -> "partitions"));
+ });
+
+ assertTrue(stats.getSchemaReadMs().isPresent());
+ assertTrue(stats.getPartitionScanMs().isPresent());
+ assertTrue(stats.getRemainingMs().isPresent());
+ }
+
+ @Test
+ void testTimedStepsThatFailAreStillRecorded() {
+ HoodieHiveSyncStats stats = new HoodieHiveSyncStats(true);
+ assertThrows(IllegalStateException.class, () -> stats.timeTotal(() ->
stats.timePartitionScan(() -> {
+ throw new IllegalStateException("listing fails");
+ })));
+
+ assertTrue(stats.getPartitionScanMs().isPresent());
+ assertTrue(stats.getRemainingMs().isPresent());
+ }
+
+ @Test
+ void testDisabledStatsRunTheStepsAndRecordNothing() {
+ HoodieHiveSyncStats stats = new HoodieHiveSyncStats(false);
+ stats.timeTotal(() -> {
+ assertEquals("schema", stats.timeSchemaRead(() -> "schema"));
+ assertEquals("partitions", stats.timePartitionScan(() -> "partitions"));
+ });
+ stats.recordPartitionsAdded(3);
+ stats.markSchemaEvolved();
+
+ assertFalse(stats.getSchemaReadMs().isPresent());
+ assertFalse(stats.getPartitionScanMs().isPresent());
+ assertFalse(stats.getRemainingMs().isPresent());
+ assertEquals(0, stats.getPartitionsAdded());
+ assertFalse(stats.isSchemaEvolved());
+ }
+}
