[flink] When reading DataSplit, calculate fetch event time lag with e…

…arliest file creation time instead of latest (#3952)
apache · Aug 13, 2024 · 7dd2edb · 7dd2edb
1 parent c4762cc
commit 7dd2edb
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 2 deletions.
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java
@@ -112,6 +112,10 @@ public OptionalLong latestFileCreationEpochMillis() {
         return this.dataFiles.stream().mapToLong(DataFileMeta::creationTimeEpochMillis).max();
     }
 
+    public OptionalLong earliestFileCreationEpochMillis() {
+        return this.dataFiles.stream().mapToLong(DataFileMeta::creationTimeEpochMillis).min();
+    }
+
     @Override
     public long rowCount() {
         long rowCount = 0;

diff --git a/...flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java b/...flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java
@@ -181,7 +181,7 @@ private void checkSplitOrStartNext() throws IOException {
         if (nextSplit.split() instanceof DataSplit) {
             long eventTime =
                     ((DataSplit) nextSplit.split())
-                            .latestFileCreationEpochMillis()
+                            .earliestFileCreationEpochMillis()
                             .orElse(FileStoreSourceReaderMetrics.UNDEFINED);
             metrics.recordSnapshotUpdate(eventTime);
         }

diff --git a/...imon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java b/...imon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java
@@ -100,7 +100,7 @@ public void processElement(StreamRecord<Split> record) throws Exception {
         // update metric when reading a new split
         long eventTime =
                 ((DataSplit) split)
-                        .latestFileCreationEpochMillis()
+                        .earliestFileCreationEpochMillis()
                         .orElse(FileStoreSourceReaderMetrics.UNDEFINED);
         sourceReaderMetrics.recordSnapshotUpdate(eventTime);