[lake/tiering] add total bytes read tracking and metrics for tiering jobs

beryllw · beryllw · commit 71361ee7b013 · 2026-03-25T17:54:24.000+08:00
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/BatchScanner.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/BatchScanner.java
@@ -47,6 +47,18 @@ public interface BatchScanner extends Closeable {
     @Nullable
     CloseableIterator<InternalRow> pollBatch(Duration timeout) throws IOException;
 
+    /**
+     * Returns the total bytes read from the underlying storage.
+     *
+     * <p>For snapshot scanners, this is the sum of value lengths read from RocksDB. For other
+     * scanners, this may return 0 if not applicable.
+     *
+     * @return total bytes read, or 0 if not applicable
+     */
+    default long getTotalBytesRead() {
+        throw new UnsupportedOperationException("Not supported.");
+    }
+
     /** Closes the scanner and should release all resources. */
     @Override
     void close() throws IOException;
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotBatchScanner.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotBatchScanner.java
@@ -180,6 +180,11 @@ public void close() {
         }
     }
 
+    @Override
+    public long getTotalBytesRead() {
+        return snapshotFilesReader != null ? snapshotFilesReader.getTotalBytesRead() : 0;
+    }
+
     private void initReaderAsynchronously() {
         CompletableFuture.runAsync(
                 () ->
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/SnapshotFilesReader.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/SnapshotFilesReader.java
@@ -74,6 +74,9 @@ class SnapshotFilesReader implements CloseableIterator<InternalRow> {
 
     private final CloseableRegistry closeableRegistry;
 
+    /** Total bytes read from RocksDB (sum of key + value lengths). */
+    private long totalBytesRead = 0;
+
     SnapshotFilesReader(
             KvFormat kvFormat,
             Path rocksDbPath,
@@ -160,9 +163,16 @@ public boolean hasNext() {
         return !isClose && rocksIteratorWrapper.isValid();
     }
 
+    /** Returns the total bytes read from RocksDB (sum of value lengths). */
+    public long getTotalBytesRead() {
+        return totalBytesRead;
+    }
+
     @Override
     public InternalRow next() {
         byte[] value = rocksIteratorWrapper.value();
+        // accumulate bytes read (key + value)
+        totalBytesRead += value.length;
         rocksIteratorWrapper.next();
 
         BinaryValue originValue = valueDecoder.decodeValue(value);
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
@@ -83,8 +83,9 @@ public LogFetchCollector(
      * @throws LogOffsetOutOfRangeException If there is OffsetOutOfRange error in fetchResponse and
      *     the defaultResetPolicy is NONE
      */
-    public Map<TableBucket, List<ScanRecord>> collectFetch(final LogFetchBuffer logFetchBuffer) {
+    public ScanRecords collectFetch(final LogFetchBuffer logFetchBuffer) {
         Map<TableBucket, List<ScanRecord>> fetched = new HashMap<>();
+        long totalSizeInBytes = 0;
         int recordsRemaining = maxPollRecords;
 
         try {
@@ -109,6 +110,8 @@ public Map<TableBucket, List<ScanRecord>> collectFetch(final LogFetchBuffer logF
                             }
                             throw e;
                         }
+                        // count bytes the first time we see this CompletedFetch
+                        totalSizeInBytes += completedFetch.sizeInBytes;
                     } else {
                         logFetchBuffer.setNextInLineFetch(completedFetch);
                     }
@@ -143,7 +146,7 @@ public Map<TableBucket, List<ScanRecord>> collectFetch(final LogFetchBuffer logF
             }
         }
 
-        return fetched;
+        return new ScanRecords(fetched, totalSizeInBytes);
     }
 
     private List<ScanRecord> fetchRecords(CompletedFetch nextInLineFetch, int maxRecords) {
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
@@ -22,7 +22,6 @@
 import org.apache.fluss.client.metadata.MetadataUpdater;
 import org.apache.fluss.client.metrics.ScannerMetricGroup;
 import org.apache.fluss.client.table.scanner.RemoteFileDownloader;
-import org.apache.fluss.client.table.scanner.ScanRecord;
 import org.apache.fluss.cluster.BucketLocation;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
@@ -161,7 +160,7 @@ public boolean hasAvailableFetches() {
         return !logFetchBuffer.isEmpty();
     }
 
-    public Map<TableBucket, List<ScanRecord>> collectFetch() {
+    public ScanRecords collectFetch() {
         return logFetchCollector.collectFetch(logFetchBuffer);
     }
 
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScannerImpl.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScannerImpl.java
@@ -21,7 +21,6 @@
 import org.apache.fluss.client.metadata.MetadataUpdater;
 import org.apache.fluss.client.metrics.ScannerMetricGroup;
 import org.apache.fluss.client.table.scanner.RemoteFileDownloader;
-import org.apache.fluss.client.table.scanner.ScanRecord;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.WakeupException;
 import org.apache.fluss.metadata.SchemaGetter;
@@ -41,8 +40,6 @@
 import java.time.Duration;
 import java.util.Collections;
 import java.util.ConcurrentModificationException;
-import java.util.List;
-import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -141,25 +138,25 @@ public ScanRecords poll(Duration timeout) {
             long timeoutNanos = timeout.toNanos();
             long startNanos = System.nanoTime();
             do {
-                Map<TableBucket, List<ScanRecord>> fetchRecords = pollForFetches();
-                if (fetchRecords.isEmpty()) {
+                ScanRecords scanRecords = pollForFetches();
+                if (scanRecords.isEmpty()) {
                     try {
                         if (!logFetcher.awaitNotEmpty(startNanos + timeoutNanos)) {
                             // logFetcher waits for the timeout and no data in buffer,
                             // so we return empty
-                            return new ScanRecords(fetchRecords);
+                            return scanRecords;
                         }
                     } catch (WakeupException e) {
                         // wakeup() is called, we need to return empty
-                        return new ScanRecords(fetchRecords);
+                        return scanRecords;
                     }
                 } else {
                     // before returning the fetched records, we can send off the next round of
                     // fetches and avoid block waiting for their responses to enable pipelining
                     // while the user is handling the fetched records.
                     logFetcher.sendFetches();
 
-                    return new ScanRecords(fetchRecords);
+                    return scanRecords;
                 }
             } while (System.nanoTime() - startNanos < timeoutNanos);
 
@@ -247,10 +244,10 @@ public void wakeup() {
         logFetcher.wakeup();
     }
 
-    private Map<TableBucket, List<ScanRecord>> pollForFetches() {
-        Map<TableBucket, List<ScanRecord>> fetchedRecords = logFetcher.collectFetch();
-        if (!fetchedRecords.isEmpty()) {
-            return fetchedRecords;
+    private ScanRecords pollForFetches() {
+        ScanRecords scanRecords = logFetcher.collectFetch();
+        if (!scanRecords.isEmpty()) {
+            return scanRecords;
         }
 
         // send any new fetches (won't resend pending fetches).
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ScanRecords.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ScanRecords.java
@@ -40,9 +40,24 @@ public class ScanRecords implements Iterable<ScanRecord> {
     public static final ScanRecords EMPTY = new ScanRecords(Collections.emptyMap());
 
     private final Map<TableBucket, List<ScanRecord>> records;
+    private final long sizeInBytes;
 
     public ScanRecords(Map<TableBucket, List<ScanRecord>> records) {
+        this(records, 0L);
+    }
+
+    public ScanRecords(Map<TableBucket, List<ScanRecord>> records, long sizeInBytes) {
         this.records = records;
+        this.sizeInBytes = sizeInBytes;
+    }
+
+    /**
+     * Total size in bytes of the fetch response that produced this record set. For log records,
+     * this reflects the actual bytes received from the server (including batch headers). Returns 0
+     * if size information is unavailable (e.g. for empty or synthetic records).
+     */
+    public long sizeInBytes() {
+        return sizeInBytes;
     }
 
     /**
diff --git a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
@@ -280,4 +280,12 @@ public class MetricNames {
     public static final String NETTY_NUM_ALLOCATIONS_PER_SECONDS = "numAllocationsPerSecond";
     public static final String NETTY_NUM_HUGE_ALLOCATIONS_PER_SECONDS =
             "numHugeAllocationsPerSecond";
+
+    // --------------------------------------------------------------------------------------------
+    // metrics for tiering service
+    // --------------------------------------------------------------------------------------------
+
+    // for lake tiering metrics - operator level
+    public static final String TIERING_SERVICE_READ_BYTES = "readBytes";
+    public static final String TIERING_SERVICE_READ_BYTES_RATE = "readBytesPerSecond";
 }
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/BoundedSplitReader.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/BoundedSplitReader.java
@@ -134,6 +134,11 @@ public void close() throws Exception {
         splitScanner.close();
     }
 
+    /** Returns the total bytes read from the underlying scanner. */
+    public long getTotalBytesRead() {
+        return splitScanner.getTotalBytesRead();
+    }
+
     private static class ScanRecordBatch implements CloseableIterator<ScanRecord> {
         private final CloseableIterator<InternalRow> rowIterator;
         private int currentSplitIndex;
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceReader.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceReader.java
@@ -22,6 +22,7 @@
 import org.apache.fluss.client.Connection;
 import org.apache.fluss.flink.adapter.SingleThreadMultiplexSourceReaderBaseAdapter;
 import org.apache.fluss.flink.tiering.event.TieringReachMaxDurationEvent;
+import org.apache.fluss.flink.tiering.source.metrics.TieringReaderMetrics;
 import org.apache.fluss.flink.tiering.source.split.TieringSplit;
 import org.apache.fluss.flink.tiering.source.state.TieringSplitState;
 import org.apache.fluss.lake.writer.LakeTieringFactory;
@@ -73,17 +74,31 @@ public TieringSourceReader(
             Duration pollTimeout) {
         super(
                 elementsQueue,
-                new TieringSourceFetcherManager<>(
-                        elementsQueue,
-                        () -> new TieringSplitReader<>(connection, lakeTieringFactory, pollTimeout),
-                        context.getConfiguration(),
-                        (ignore) -> {}),
+                createFetcherManager(
+                        elementsQueue, context, connection, lakeTieringFactory, pollTimeout),
                 new TableBucketWriteResultEmitter<>(),
                 context.getConfiguration(),
                 context);
         this.connection = connection;
     }
 
+    private static <WriteResult> TieringSourceFetcherManager<WriteResult> createFetcherManager(
+            FutureCompletingBlockingQueue<RecordsWithSplitIds<TableBucketWriteResult<WriteResult>>>
+                    elementsQueue,
+            SourceReaderContext context,
+            Connection connection,
+            LakeTieringFactory<WriteResult, ?> lakeTieringFactory,
+            Duration pollTimeout) {
+        TieringReaderMetrics tieringReaderMetrics = new TieringReaderMetrics(context.metricGroup());
+        return new TieringSourceFetcherManager<>(
+                elementsQueue,
+                () ->
+                        new TieringSplitReader<>(
+                                connection, lakeTieringFactory, pollTimeout, tieringReaderMetrics),
+                context.getConfiguration(),
+                (ignore) -> {});
+    }
+
     @Override
     public void start() {
         // we request a split only if we did not get splits during the checkpoint restore
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java
@@ -25,6 +25,7 @@
 import org.apache.fluss.client.table.scanner.log.ScanRecords;
 import org.apache.fluss.flink.source.reader.BoundedSplitReader;
 import org.apache.fluss.flink.source.reader.RecordAndPos;
+import org.apache.fluss.flink.tiering.source.metrics.TieringReaderMetrics;
 import org.apache.fluss.flink.tiering.source.split.TieringLogSplit;
 import org.apache.fluss.flink.tiering.source.split.TieringSnapshotSplit;
 import org.apache.fluss.flink.tiering.source.split.TieringSplit;
@@ -106,16 +107,27 @@ public class TieringSplitReader<WriteResult>
 
     private final Set<TieringSplit> currentEmptySplits;
 
+    @Nullable private final TieringReaderMetrics tieringReaderMetrics;
+
+    /** Creates a split reader without metrics. Use the 4-argument constructor in production. */
     public TieringSplitReader(
             Connection connection, LakeTieringFactory<WriteResult, ?> lakeTieringFactory) {
-        this(connection, lakeTieringFactory, DEFAULT_POLL_TIMEOUT);
+        this(connection, lakeTieringFactory, DEFAULT_POLL_TIMEOUT, null);
     }
 
     @VisibleForTesting
     protected TieringSplitReader(
             Connection connection,
             LakeTieringFactory<WriteResult, ?> lakeTieringFactory,
             Duration pollTimeout) {
+        this(connection, lakeTieringFactory, pollTimeout, null);
+    }
+
+    protected TieringSplitReader(
+            Connection connection,
+            LakeTieringFactory<WriteResult, ?> lakeTieringFactory,
+            Duration pollTimeout,
+            @Nullable TieringReaderMetrics tieringReaderMetrics) {
         this.lakeTieringFactory = lakeTieringFactory;
         // owned by TieringSourceReader
         this.connection = connection;
@@ -129,6 +141,7 @@ protected TieringSplitReader(
         this.currentPendingSnapshotSplits = new ArrayDeque<>();
         this.reachTieringMaxDurationTables = new HashSet<>();
         this.pollTimeout = pollTimeout;
+        this.tieringReaderMetrics = tieringReaderMetrics;
     }
 
     @Override
@@ -345,6 +358,11 @@ private void mayCreateLogScanner() {
 
     private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> forLogRecords(
             ScanRecords scanRecords) throws IOException {
+        // record bytes read from Fluss for log records
+        if (tieringReaderMetrics != null) {
+            tieringReaderMetrics.recordBytesRead(scanRecords.sizeInBytes());
+        }
+
         Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults = new HashMap<>();
         Map<TableBucket, String> finishedSplitIds = new HashMap<>();
         LOG.info("for log records to tier table {}.", currentTableId);
@@ -507,6 +525,13 @@ private TableBucketWriteResultWithSplitIds forSnapshotSplitRecords(
             lakeWriter.write(scanRecord);
         }
         recordIterator.close();
+
+        // record bytes read from Fluss for snapshot records
+        // get the actual bytes from the underlying scanner
+        if (tieringReaderMetrics != null && currentSnapshotSplitReader != null) {
+            tieringReaderMetrics.recordBytesRead(currentSnapshotSplitReader.getTotalBytesRead());
+        }
+
         return emptyTableBucketWriteResultWithSplitIds();
     }
 
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/metrics/TieringReaderMetrics.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/metrics/TieringReaderMetrics.java

Original file line number	Diff line number	Diff line change
`@@ -180,6 +180,11 @@ public void close() {`
`180`	`180`	`}`
`181`	`181`	`}`
`182`	`182`
	`183`	`+ @Override`
	`184`	`+ public long getTotalBytesRead() {`
	`185`	`+ return snapshotFilesReader != null ? snapshotFilesReader.getTotalBytesRead() : 0;`
	`186`	`+ }`
	`187`	`+`
`183`	`188`	`private void initReaderAsynchronously() {`
`184`	`189`	`CompletableFuture.runAsync(`
`185`	`190`	`() ->`
Original file line number	Diff line number	Diff line change
`@@ -83,8 +83,9 @@ public LogFetchCollector(`
`83`	`83`	`* @throws LogOffsetOutOfRangeException If there is OffsetOutOfRange error in fetchResponse and`
`84`	`84`	`* the defaultResetPolicy is NONE`
`85`	`85`	`*/`
`86`		`- public Map<TableBucket, List<ScanRecord>> collectFetch(final LogFetchBuffer logFetchBuffer) {`
	`86`	`+ public ScanRecords collectFetch(final LogFetchBuffer logFetchBuffer) {`
`87`	`87`	`Map<TableBucket, List<ScanRecord>> fetched = new HashMap<>();`
	`88`	`+ long totalSizeInBytes = 0;`
`88`	`89`	`int recordsRemaining = maxPollRecords;`
`89`	`90`
`90`	`91`	`try {`
`@@ -109,6 +110,8 @@ public Map<TableBucket, List<ScanRecord>> collectFetch(final LogFetchBuffer logF`
`109`	`110`	`}`
`110`	`111`	`throw e;`
`111`	`112`	`}`
	`113`	`+ // count bytes the first time we see this CompletedFetch`
	`114`	`+ totalSizeInBytes += completedFetch.sizeInBytes;`
`112`	`115`	`} else {`
`113`	`116`	`logFetchBuffer.setNextInLineFetch(completedFetch);`
`114`	`117`	`}`
`@@ -143,7 +146,7 @@ public Map<TableBucket, List<ScanRecord>> collectFetch(final LogFetchBuffer logF`
`143`	`146`	`}`
`144`	`147`	`}`
`145`	`148`
`146`		`- return fetched;`
	`149`	`+ return new ScanRecords(fetched, totalSizeInBytes);`
`147`	`150`	`}`
`148`	`151`
`149`	`152`	`private List<ScanRecord> fetchRecords(CompletedFetch nextInLineFetch, int maxRecords) {`
Original file line number	Diff line number	Diff line change
`@@ -22,7 +22,6 @@`
`22`	`22`	`import org.apache.fluss.client.metadata.MetadataUpdater;`
`23`	`23`	`import org.apache.fluss.client.metrics.ScannerMetricGroup;`
`24`	`24`	`import org.apache.fluss.client.table.scanner.RemoteFileDownloader;`
`25`		`-import org.apache.fluss.client.table.scanner.ScanRecord;`
`26`	`25`	`import org.apache.fluss.cluster.BucketLocation;`
`27`	`26`	`import org.apache.fluss.config.ConfigOptions;`
`28`	`27`	`import org.apache.fluss.config.Configuration;`
`@@ -161,7 +160,7 @@ public boolean hasAvailableFetches() {`
`161`	`160`	`return !logFetchBuffer.isEmpty();`
`162`	`161`	`}`
`163`	`162`
`164`		`- public Map<TableBucket, List<ScanRecord>> collectFetch() {`
	`163`	`+ public ScanRecords collectFetch() {`
`165`	`164`	`return logFetchCollector.collectFetch(logFetchBuffer);`
`166`	`165`	`}`
`167`	`166`