[metrics] Fix pendingRecords metric to return actual row count for primary key tables when lake tiering hasn't started

Author: beryllw

fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java

@@ -222,11 +222,13 @@ public class MetricNames {
     // metrics for table bucket
     // --------------------------------------------------------------------------------------------
 
+    // for tablet
+    public static final String LAKE_PENDING_RECORDS = "pendingRecords";
+
     // for log tablet
     public static final String LOG_NUM_SEGMENTS = "numSegments";
     public static final String LOG_END_OFFSET = "endOffset";
     public static final String REMOTE_LOG_SIZE = "size";
-    public static final String LOG_LAKE_PENDING_RECORDS = "pendingRecords";
     public static final String LOG_LAKE_TIMESTAMP_LAG = "timestampLag";
 
     // for logic storage

fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java

@@ -23,6 +23,7 @@
 import org.apache.fluss.config.TableConfig;
 import org.apache.fluss.exception.FencedLeaderEpochException;
 import org.apache.fluss.exception.InvalidColumnProjectionException;
+import org.apache.fluss.exception.InvalidTableException;
 import org.apache.fluss.exception.InvalidTimestampException;
 import org.apache.fluss.exception.InvalidUpdateVersionException;
 import org.apache.fluss.exception.KvStorageException;
@@ -549,11 +550,20 @@ private void onBecomeNewLeader() {
     private void registerLakeTieringMetrics() {
         lakeTieringMetricGroup = bucketMetricGroup.addGroup("lakeTiering");
         lakeTieringMetricGroup.gauge(
-                MetricNames.LOG_LAKE_PENDING_RECORDS,
-                () ->
-                        getLakeLogEndOffset() < 0L
-                                ? getLogHighWatermark() - getLogStartOffset()
-                                : getLogHighWatermark() - getLakeLogEndOffset());
+                MetricNames.LAKE_PENDING_RECORDS,
+                () -> {
+                    long lakeLogEndOffset = getLakeLogEndOffset();
+                    if (lakeLogEndOffset < 0L) {
+                        try {
+                            return getRowCount();
+                        } catch (InvalidTableException e) {
+                            // WAL mode or v0.9 old table: row count disabled,
+                            // fall back to log-level pending count
+                            return getLogHighWatermark() - getLogStartOffset();
+                        }
+                    }
+                    return getLogHighWatermark() - lakeLogEndOffset;
+                });
         lakeTieringMetricGroup.gauge(
                 MetricNames.LOG_LAKE_TIMESTAMP_LAG,
                 () ->

website/docs/maintenance/observability/monitor-metrics.md

@@ -855,7 +855,7 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM
     </tr>
      <tr>
       <td>timestampLag</td>
-      <td>The timestamp lag between local log and remote log for this table bucket in milliseconds.</td>
+      <td>The timestamp lag between the latest log record and the latest record already tiered to the lake for this table bucket, in milliseconds.</td>
       <td>Gauge</td>
     </tr>
     <tr>