[BugFix] Use snapshot id for PaimonFilter to avoid cache coherency pr…

…oblem Signed-off-by: Jiao Mingye <mxdzs0612@gmail.com>
mxdzs0612 · Feb 11, 2025 · 3a4510f · 3a4510f
1 parent 18f1714
commit 3a4510f
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 67 deletions.
diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/paimon/PaimonFilter.java b/fe/fe-core/src/main/java/com/starrocks/connector/paimon/PaimonFilter.java
diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/paimon/PaimonMetadata.java b/fe/fe-core/src/main/java/com/starrocks/connector/paimon/PaimonMetadata.java
@@ -29,6 +29,7 @@
 import com.starrocks.connector.GetRemoteFilesParams;
 import com.starrocks.connector.HdfsEnvironment;
 import com.starrocks.connector.PartitionInfo;
+import com.starrocks.connector.PredicateSearchKey;
 import com.starrocks.connector.RemoteFileDesc;
 import com.starrocks.connector.RemoteFileInfo;
 import com.starrocks.connector.TableVersionRange;
@@ -78,7 +79,7 @@ public class PaimonMetadata implements ConnectorMetadata {
     private final String catalogName;
     private final Map<Identifier, Table> tables = new ConcurrentHashMap<>();
     private final Map<String, Database> databases = new ConcurrentHashMap<>();
-    private final Map<PaimonFilter, PaimonSplitsInfo> paimonSplits = new ConcurrentHashMap<>();
+    private final Map<PredicateSearchKey, PaimonSplitsInfo> paimonSplits = new ConcurrentHashMap<>();
     private final Map<String, Long> partitionInfos = new ConcurrentHashMap<>();
     private final ConnectorProperties properties;
 
@@ -249,9 +250,12 @@ public boolean tableExists(String dbName, String tableName) {
     public List<RemoteFileInfo> getRemoteFiles(Table table, GetRemoteFilesParams params) {
         RemoteFileInfo remoteFileInfo = new RemoteFileInfo();
         PaimonTable paimonTable = (PaimonTable) table;
-        PaimonFilter filter =
-                new PaimonFilter(paimonTable.getCatalogDBName(), paimonTable.getCatalogTableName(), params.getPredicate(),
-                        params.getFieldNames());
+        long latestSnapshotId = -1L;
+        if (paimonTable.getNativeTable().latestSnapshotId().isPresent()) {
+            latestSnapshotId = paimonTable.getNativeTable().latestSnapshotId().getAsLong();
+        }
+        PredicateSearchKey filter = PredicateSearchKey.of(paimonTable.getCatalogDBName(),
+                paimonTable.getCatalogTableName(), latestSnapshotId, params.getPredicate());
         if (!paimonSplits.containsKey(filter)) {
             ReadBuilder readBuilder = paimonTable.getNativeTable().newReadBuilder();
             int[] projected =