add some metrics for upsert table preloading

pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java

@@ -53,6 +53,7 @@ public enum ServerMeter implements AbstractMetrics.Meter {
   DELETED_KEYS_TTL_PRIMARY_KEYS_REMOVED("rows", false),
   METADATA_TTL_PRIMARY_KEYS_REMOVED("rows", false),
   UPSERT_MISSED_VALID_DOC_ID_SNAPSHOT_COUNT("segments", false),
+  UPSERT_PRELOAD_FAILURE("count", false),
   ROWS_WITH_ERRORS("rows", false),
   LLC_CONTROLLER_RESPONSE_NOT_SENT("messages", true),
   LLC_CONTROLLER_RESPONSE_COMMIT("messages", true),

pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerTimer.java

@@ -49,6 +49,8 @@ public enum ServerTimer implements AbstractMetrics.Timer {
   TOTAL_CPU_TIME_NS("nanoseconds", false, "Total query cost (thread cpu time + system "
       + "activities cpu time + response serialization cpu time) for query processing on server."),
 
+  UPSERT_PRELOAD_TIME_MS("milliseconds", false,
+      "Total time taken to preload a table partition of an upsert table with upsert snapshot"),
   UPSERT_REMOVE_EXPIRED_PRIMARY_KEYS_TIME_MS("milliseconds", false,
       "Total time taken to delete expired primary keys based on metadataTTL or deletedKeysTTL"),
   UPSERT_SNAPSHOT_TIME_MS("milliseconds", false, "Total time taken to take upsert table snapshot");

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java

@@ -199,14 +199,19 @@ public void preloadSegments(IndexLoadingConfig indexLoadingConfig) {
         return;
       }
       // From now on, the _isPreloading flag is true until the segments are preloaded.
+      long startTime = System.currentTimeMillis();
       doPreloadSegments(tableDataManager, indexLoadingConfig, helixManager, segmentPreloadExecutor);
+      long duration = System.currentTimeMillis() - startTime;
+      _serverMetrics.addTimedTableValue(_tableNameWithType, ServerTimer.UPSERT_PRELOAD_TIME_MS, duration,
+          TimeUnit.MILLISECONDS);
     } catch (Exception e) {
       // Even if preloading fails, we should continue to complete the initialization, so that TableDataManager can be
       // created. Once TableDataManager is created, no more segment preloading would happen, and the normal segment
       // loading logic would be used. The segments not being preloaded successfully here would be loaded via the
       // normal segment loading logic, the one doing more costly checks on the upsert metadata.
       _logger.warn("Failed to preload segments from partition: {} of table: {}, skipping", _partitionId,
           _tableNameWithType, e);
+      _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.UPSERT_PRELOAD_FAILURE, 1);
       if (e instanceof InterruptedException) {
         // Restore the interrupted status in case the upper callers want to check.
         Thread.currentThread().interrupt();