hudi-agent commented on code in PR #20165:
URL: https://github.com/apache/hudi/pull/20165#discussion_r4291056709
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java:
##########
@@ -465,7 +467,9 @@ private void initHiveSync() {
.waitForTasksFinish(true).build();
this.hiveSyncContext = HiveSyncContext.create(conf, this.storageConf);
this.hiveSyncMetrics = new FlinkHiveSyncMetrics(context.metricGroup());
- this.hiveSyncMetrics.registerMetrics();
+ if (FormatUtils.getBooleanWithAltKeys(conf,
HiveSyncConfigHolder.HIVE_SYNC_STATS_ENABLED)) {
Review Comment:
🤖 This gate also skips registering `hiveSyncSuccessCount`,
`hiveSyncFailureCount`, `hiveSyncDurationMs` and `hiveSyncLastSuccessTimeMs`
from #20148, so by default a table that stops syncing goes back to leaving no
trace outside the JM log; would it make sense to always register those four and
gate only the breakdown metrics (init, schema read, partition scan, remaining,
partitions added, schema evolved)?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncStats.java:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.Option;
+
+/**
+ * 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 and the schema flag
describe one of them,
+ * since each gets the same partitions and the same schema.
+ *
+ * <p>After a failed sync, the stats cover the steps that ran before the
failure.
+ *
+ * <p>Unless {@code hoodie.datasource.hive_sync.stats.enabled} is set, the
sync records into
+ * disabled stats, which stay empty.
+ */
+public class HiveSyncStats {
+
+ private final boolean enabled;
+ private long totalMs = -1;
+ private long schemaReadMs = -1;
+ private long partitionScanMs = -1;
+ private int partitionsAdded = 0;
+ private boolean schemaEvolved = false;
+
+ /**
+ * @param enabled whether to keep what is recorded; disabled stats ignore it
and stay empty
+ */
+ public HiveSyncStats(boolean enabled) {
+ this.enabled = enabled;
+ }
+
+ /**
+ * 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)));
+ }
+
+ /** Number of partitions added to the metastore. */
+ 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 setTotalMs(long totalMs) {
+ if (enabled) {
Review Comment:
🤖 nit: could you avoid repeating the `if (enabled)` guard in every setter,
e.g. by making disabled stats a no-op subclass or a shared NOOP instance?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
