add bytes read tracking for log fetch operations

Author: beryllw

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/CompletedFetch.java

@@ -109,6 +109,10 @@ boolean isConsumed() {
         return isConsumed;
     }
 
+    public int getSizeInBytes() {
+        return sizeInBytes;
+    }
+
     boolean isInitialized() {
         return initialized;
     }

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java

@@ -86,6 +86,7 @@ public LogFetchCollector(
     public ScanRecords collectFetch(final LogFetchBuffer logFetchBuffer) {
         Map<TableBucket, List<ScanRecord>> fetched = new HashMap<>();
         int recordsRemaining = maxPollRecords;
+        long totalBytesRead = 0;
 
         try {
             while (recordsRemaining > 0) {
@@ -135,6 +136,11 @@ public ScanRecords collectFetch(final LogFetchBuffer logFetchBuffer) {
 
                         recordsRemaining -= records.size();
                     }
+
+                    // Only count bytes when the fetch is fully consumed
+                    if (nextInLineFetch.isConsumed()) {
+                        totalBytesRead += nextInLineFetch.getSizeInBytes();
+                    }
                 }
             }
         } catch (FetchException e) {
@@ -143,7 +149,7 @@ public ScanRecords collectFetch(final LogFetchBuffer logFetchBuffer) {
             }
         }
 
-        return new ScanRecords(fetched);
+        return new ScanRecords(fetched, totalBytesRead);
     }
 
     private List<ScanRecord> fetchRecords(CompletedFetch nextInLineFetch, int maxRecords) {

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ScanRecords.java

@@ -37,12 +37,27 @@
  */
 @PublicEvolving
 public class ScanRecords implements Iterable<ScanRecord> {
-    public static final ScanRecords EMPTY = new ScanRecords(Collections.emptyMap());
+    public static final ScanRecords EMPTY = new ScanRecords(Collections.emptyMap(), 0);
 
     private final Map<TableBucket, List<ScanRecord>> records;
+    private final long totalBytesRead;
 
     public ScanRecords(Map<TableBucket, List<ScanRecord>> records) {
+        this(records, 0);
+    }
+
+    public ScanRecords(Map<TableBucket, List<ScanRecord>> records, long totalBytesRead) {
         this.records = records;
+        this.totalBytesRead = totalBytesRead;
+    }
+
+    /**
+     * Get the total bytes read from the Fluss log in this batch.
+     *
+     * @return the total bytes read
+     */
+    public long getTotalBytesRead() {
+        return totalBytesRead;
     }
 
     /**

fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java

@@ -153,6 +153,41 @@ void testCollectAfterUnassign() throws Exception {
         assertThat(bucketAndRecords.buckets().size()).isEqualTo(0);
     }
 
+    @Test
+    void testTotalBytesRead() throws Exception {
+        TableBucket tb1 = new TableBucket(DATA1_TABLE_ID, 1L, 1);
+        TableBucket tb2 = new TableBucket(DATA1_TABLE_ID, 1L, 2);
+        Map<TableBucket, Long> scanBuckets = new HashMap<>();
+        scanBuckets.put(tb1, 0L);
+        scanBuckets.put(tb2, 0L);
+        logScannerStatus.assignScanBuckets(scanBuckets);
+
+        CompletedFetch completedFetch1 =
+                makeCompletedFetch(
+                        tb1,
+                        new FetchLogResultForBucket(tb1, genMemoryLogRecordsByObject(DATA1), 10L),
+                        0L);
+        CompletedFetch completedFetch2 =
+                makeCompletedFetch(
+                        tb2,
+                        new FetchLogResultForBucket(tb2, genMemoryLogRecordsByObject(DATA1), 10L),
+                        0L);
+
+        logFetchBuffer.add(completedFetch1);
+        logFetchBuffer.add(completedFetch2);
+
+        ScanRecords scanRecords = logFetchCollector.collectFetch(logFetchBuffer);
+
+        // Both fetches should be fully consumed
+        assertThat(completedFetch1.isConsumed()).isTrue();
+        assertThat(completedFetch2.isConsumed()).isTrue();
+
+        // totalBytesRead should be the sum of both completed fetches
+        long expectedBytes = completedFetch1.getSizeInBytes() + completedFetch2.getSizeInBytes();
+        assertThat(scanRecords.getTotalBytesRead()).isEqualTo(expectedBytes);
+        assertThat(scanRecords.getTotalBytesRead()).isGreaterThan(0);
+    }
+
     private DefaultCompletedFetch makeCompletedFetch(
             TableBucket tableBucket, FetchLogResultForBucket resultForBucket, long offset) {
         return new DefaultCompletedFetch(

fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/ScanRecordsTest.java

@@ -57,4 +57,18 @@ void iterator() {
         }
         assertThat(c).isEqualTo(4);
     }
+
+    @Test
+    void testTotalBytesRead() {
+        Map<TableBucket, List<ScanRecord>> records = new LinkedHashMap<>();
+        long tableId = 0;
+        ScanRecord record1 = new ScanRecord(0L, 1000L, ChangeType.INSERT, row(1, "a"));
+        records.put(new TableBucket(tableId, 0), Arrays.asList(record1));
+
+        // New constructor carries totalBytesRead
+        assertThat(new ScanRecords(records, 1024L).getTotalBytesRead()).isEqualTo(1024L);
+
+        // Old constructor defaults to 0 for backward compatibility
+        assertThat(new ScanRecords(records).getTotalBytesRead()).isEqualTo(0L);
+    }
 }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java

@@ -34,6 +34,8 @@
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.BinaryRow;
+import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.utils.CloseableIterator;
 
 import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
@@ -355,6 +357,9 @@ private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> forLogRecords(
         Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults = new HashMap<>();
         Map<TableBucket, String> finishedSplitIds = new HashMap<>();
         LOG.info("for log records to tier table {}.", currentTableId);
+
+        // Report bytes read from Fluss log records
+        tieringMetrics.recordBytesRead(scanRecords.getTotalBytesRead());
         for (TableBucket bucket : scanRecords.buckets()) {
             LOG.info("tiering table bucket {}.", bucket);
             List<ScanRecord> bucketScanRecords = scanRecords.records(bucket);
@@ -443,7 +448,6 @@ private TableBucketWriteResult<WriteResult> completeLakeWriter(
         WriteResult writeResult = null;
         if (lakeWriter != null) {
             writeResult = lakeWriter.complete();
-            tieringMetrics.recordBytesWritten(lakeWriter.getBytesWritten());
             lakeWriter.close();
         }
         return toTableBucketWriteResult(
@@ -510,10 +514,15 @@ private TableBucketWriteResultWithSplitIds forSnapshotSplitRecords(
         LakeWriter<WriteResult> lakeWriter =
                 getOrCreateLakeWriter(
                         bucket, checkNotNull(currentSnapshotSplit).getPartitionName());
+        long bytesRead = 0;
         while (recordIterator.hasNext()) {
             ScanRecord scanRecord = recordIterator.next().record();
             lakeWriter.write(scanRecord);
+            InternalRow row = scanRecord.getRow();
+            // Snapshot path always produces BinaryRow (CompactedRow/IndexedRow).
+            bytesRead += ((BinaryRow) row).getSizeInBytes();
         }
+        tieringMetrics.recordBytesRead(bytesRead);
         recordIterator.close();
         return emptyTableBucketWriteResultWithSplitIds();
     }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/metrics/TieringMetrics.java

@@ -34,10 +34,10 @@
  * <p>The following metrics are available:
  *
  * <ul>
- *   <li>{@code fluss.tieringService.writtenBytes} - Counter: cumulative bytes written to the lake
- *       since the job started (physical file sizes).
- *   <li>{@code fluss.tieringService.writtenBytesPerSecond} - Meter: write bytes-per-second rate
- *       derived from the counter using a 60-second sliding window.
+ *   <li>{@code fluss.tieringService.writtenBytes} - Counter: cumulative bytes read from Fluss
+ *       records since the job started (actual Fluss records size).
+ *   <li>{@code fluss.tieringService.writtenBytesPerSecond} - Meter: bytes-per-second rate derived
+ *       from the counter using a 60-second sliding window.
  * </ul>
  */
 @Internal
@@ -61,8 +61,8 @@ public TieringMetrics(SourceReaderMetricGroup sourceReaderMetricGroup) {
                 MetricNames.TIERING_SERVICE_WRITTEN_BYTES_RATE, new MeterView(writtenBytesCounter));
     }
 
-    /** Records bytes written to the lake. Called once per bucket completion. */
-    public void recordBytesWritten(long bytes) {
+    /** Records bytes read from Fluss records. Called per batch or record processing. */
+    public void recordBytesRead(long bytes) {
         writtenBytesCounter.inc(bytes);
     }
 }

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringMetricsTest.java

@@ -0,0 +1,53 @@
+/*
+ * 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.fluss.flink.tiering.source;
+
+import org.apache.fluss.flink.tiering.source.metrics.TieringMetrics;
+import org.apache.fluss.metrics.MetricNames;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.testutils.MetricListener;
+import org.apache.flink.runtime.metrics.groups.InternalSourceReaderMetricGroup;
+import org.junit.jupiter.api.Test;
+
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link TieringMetrics}. */
+class TieringMetricsTest {
+
+    @Test
+    void testRecordBytesRead() {
+        MetricListener metricListener = new MetricListener();
+        TieringMetrics tieringMetrics =
+                new TieringMetrics(
+                        InternalSourceReaderMetricGroup.mock(metricListener.getMetricGroup()));
+
+        tieringMetrics.recordBytesRead(100);
+        tieringMetrics.recordBytesRead(200);
+
+        Optional<Counter> counter =
+                metricListener.getCounter(
+                        TieringMetrics.FLUSS_METRIC_GROUP,
+                        TieringMetrics.TIERING_SERVICE_GROUP,
+                        MetricNames.TIERING_SERVICE_WRITTEN_BYTES);
+        assertThat(counter).isPresent();
+        assertThat(counter.get().getCount()).isEqualTo(300);
+    }
+}