[core] Pass file ser version to LookupSerializerFactory to let it deserialize from old version (#6677)

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/mergetree/LookupFile.java

@@ -50,6 +50,7 @@ public class LookupFile {
     private final File localFile;
     private final int level;
     private final long schemaId;
+    private final String serVersion;
     private final LookupStoreReader reader;
     private final Runnable callback;
 
@@ -58,10 +59,16 @@ public class LookupFile {
     private boolean isClosed = false;
 
     public LookupFile(
-            File localFile, int level, long schemaId, LookupStoreReader reader, Runnable callback) {
+            File localFile,
+            int level,
+            long schemaId,
+            String serVersion,
+            LookupStoreReader reader,
+            Runnable callback) {
         this.localFile = localFile;
         this.level = level;
         this.schemaId = schemaId;
+        this.serVersion = serVersion;
         this.reader = reader;
         this.callback = callback;
     }
@@ -74,6 +81,10 @@ public long schemaId() {
         return schemaId;
     }
 
+    public String serVersion() {
+        return serVersion;
+    }
+
     @Nullable
     public byte[] get(byte[] key) throws IOException {
         checkArgument(!isClosed);

paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java

@@ -34,6 +34,7 @@
 import org.apache.paimon.utils.BloomFilter;
 import org.apache.paimon.utils.FileIOUtils;
 import org.apache.paimon.utils.IOFunction;
+import org.apache.paimon.utils.Pair;
 
 import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache;
 
@@ -45,6 +46,7 @@
 import java.util.Comparator;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.concurrent.ConcurrentHashMap;
@@ -68,8 +70,7 @@ public class LookupLevels<T> implements Levels.DropFileCallback, Closeable {
     private final Function<Long, BloomFilter.Builder> bfGenerator;
     private final Cache<String, LookupFile> lookupFileCache;
     private final Set<String> ownCachedFiles;
-    private final String remoteSstSuffix;
-    private final Map<Long, PersistProcessor<T>> schemaIdToProcessors;
+    private final Map<Pair<Long, String>, PersistProcessor<T>> schemaIdAndSerVersionToProcessors;
 
     @Nullable private RemoteFileDownloader remoteFileDownloader;
 
@@ -99,13 +100,7 @@ public LookupLevels(
         this.bfGenerator = bfGenerator;
         this.lookupFileCache = lookupFileCache;
         this.ownCachedFiles = new HashSet<>();
-        this.remoteSstSuffix =
-                "."
-                        + processorFactory.identifier()
-                        + "."
-                        + serializerFactory.identifier()
-                        + REMOTE_LOOKUP_FILE_SUFFIX;
-        this.schemaIdToProcessors = new ConcurrentHashMap<>();
+        this.schemaIdAndSerVersionToProcessors = new ConcurrentHashMap<>();
         levels.addDropFileCallback(this);
     }
 
@@ -170,17 +165,17 @@ private T lookup(InternalRow key, DataFileMeta file) throws IOException {
             return null;
         }
 
-        return getOrCreateProcessor(lookupFile.schemaId())
+        return getOrCreateProcessor(lookupFile.schemaId(), lookupFile.serVersion())
                 .readFromDisk(key, lookupFile.level(), valueBytes, file.fileName());
     }
 
-    private PersistProcessor<T> getOrCreateProcessor(long schemaId) {
-        return schemaIdToProcessors.computeIfAbsent(
-                schemaId,
+    private PersistProcessor<T> getOrCreateProcessor(long schemaId, String serVersion) {
+        return schemaIdAndSerVersionToProcessors.computeIfAbsent(
+                Pair.of(schemaId, serVersion),
                 id -> {
                     RowType fileSchema =
                             schemaId == currentSchemaId ? null : schemaFunction.apply(schemaId);
-                    return processorFactory.create(serializerFactory, fileSchema);
+                    return processorFactory.create(serVersion, serializerFactory, fileSchema);
                 });
     }
 
@@ -191,9 +186,12 @@ public LookupFile createLookupFile(DataFileMeta file) throws IOException {
         }
 
         long schemaId = this.currentSchemaId;
-        if (tryToDownloadRemoteSst(file, localFile)) {
+        String fileSerVersion = serializerFactory.version();
+        Optional<String> downloadSerVersion = tryToDownloadRemoteSst(file, localFile);
+        if (downloadSerVersion.isPresent()) {
             // use schema id from remote file
             schemaId = file.schemaId();
+            fileSerVersion = downloadSerVersion.get();
         } else {
             createSstFileFromDataFile(file, localFile);
         }
@@ -203,21 +201,35 @@ public LookupFile createLookupFile(DataFileMeta file) throws IOException {
                 localFile,
                 file.level(),
                 schemaId,
+                fileSerVersion,
                 lookupStoreFactory.createReader(localFile),
                 () -> ownCachedFiles.remove(file.fileName()));
     }
 
-    private boolean tryToDownloadRemoteSst(DataFileMeta file, File localFile) {
+    private Optional<String> tryToDownloadRemoteSst(DataFileMeta file, File localFile) {
         if (remoteFileDownloader == null) {
-            return false;
+            return Optional.empty();
+        }
+        Optional<RemoteSstFile> remoteSstFile = remoteSst(file);
+        if (!remoteSstFile.isPresent()) {
+            return Optional.empty();
         }
+
+        RemoteSstFile remoteSst = remoteSstFile.get();
+
         // validate schema matched, no exception here
         try {
-            getOrCreateProcessor(file.schemaId());
+            getOrCreateProcessor(file.schemaId(), remoteSst.serVersion);
         } catch (UnsupportedOperationException e) {
-            return false;
+            return Optional.empty();
+        }
+        boolean success =
+                remoteFileDownloader.tryToDownload(file, remoteSst.sstFileName, localFile);
+        if (!success) {
+            return Optional.empty();
         }
-        return remoteFileDownloader.tryToDownload(file, localFile);
+
+        return Optional.of(remoteSst.serVersion);
     }
 
     public void addLocalFile(DataFileMeta file, LookupFile lookupFile) {
@@ -229,7 +241,8 @@ private void createSstFileFromDataFile(DataFileMeta file, File localFile) throws
                         lookupStoreFactory.createWriter(
                                 localFile, bfGenerator.apply(file.rowCount()));
                 RecordReader<KeyValue> reader = fileReaderFactory.apply(file)) {
-            PersistProcessor<T> processor = getOrCreateProcessor(currentSchemaId);
+            PersistProcessor<T> processor =
+                    getOrCreateProcessor(currentSchemaId, serializerFactory.version());
             KeyValue kv;
             if (processor.withPosition()) {
                 FileRecordIterator<KeyValue> batch;
@@ -258,8 +271,39 @@ private void createSstFileFromDataFile(DataFileMeta file, File localFile) throws
         }
     }
 
-    public String remoteSstSuffix() {
-        return remoteSstSuffix;
+    public Optional<RemoteSstFile> remoteSst(DataFileMeta file) {
+        Optional<String> sstFile =
+                file.extraFiles().stream()
+                        .filter(f -> f.endsWith(REMOTE_LOOKUP_FILE_SUFFIX))
+                        .findFirst();
+        if (!sstFile.isPresent()) {
+            return Optional.empty();
+        }
+
+        String sstFileName = sstFile.get();
+        String[] split = sstFileName.split("\\.");
+        if (split.length < 3) {
+            return Optional.empty();
+        }
+
+        String processorId = split[split.length - 3];
+        if (!processorFactory.identifier().equals(processorId)) {
+            return Optional.empty();
+        }
+
+        String serVersion = split[split.length - 2];
+        return Optional.of(new RemoteSstFile(sstFileName, serVersion));
+    }
+
+    public String newRemoteSst(DataFileMeta file, long length) {
+        return file.fileName()
+                + "."
+                + length
+                + "."
+                + processorFactory.identifier()
+                + "."
+                + serializerFactory.version()
+                + REMOTE_LOOKUP_FILE_SUFFIX;
     }
 
     @Override
@@ -269,4 +313,16 @@ public void close() throws IOException {
             lookupFileCache.invalidate(cachedFile);
         }
     }
+
+    /** Remote sst file with serVersion. */
+    public static class RemoteSstFile {
+
+        private final String sstFileName;
+        private final String serVersion;
+
+        private RemoteSstFile(String sstFileName, String serVersion) {
+            this.sstFileName = sstFileName;
+            this.serVersion = serVersion;
+        }
+    }
 }

paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/DefaultLookupSerializerFactory.java

@@ -30,7 +30,7 @@
 public class DefaultLookupSerializerFactory implements LookupSerializerFactory {
 
     @Override
-    public String identifier() {
+    public String version() {
         return "v1";
     }
 
@@ -42,7 +42,10 @@ public Function<InternalRow, byte[]> createSerializer(RowType currentSchema) {
 
     @Override
     public Function<byte[], InternalRow> createDeserializer(
-            RowType currentSchema, @Nullable RowType fileSchema) {
+            String fileSerVersion, RowType currentSchema, @Nullable RowType fileSchema) {
+        if (!version().equals(fileSerVersion)) {
+            throw new UnsupportedOperationException();
+        }
         if (fileSchema != null && !fileSchema.equalsIgnoreNullable(currentSchema)) {
             throw new UnsupportedOperationException();
         }

paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/LookupSerializerFactory.java

@@ -32,12 +32,12 @@
 /** Factory to create serializer for lookup. */
 public interface LookupSerializerFactory {
 
-    String identifier();
+    String version();
 
     Function<InternalRow, byte[]> createSerializer(RowType currentSchema);
 
     Function<byte[], InternalRow> createDeserializer(
-            RowType currentSchema, @Nullable RowType fileSchema);
+            String fileSerVersion, RowType currentSchema, @Nullable RowType fileSchema);
 
     Supplier<LookupSerializerFactory> INSTANCE =
             Suppliers.memoize(

paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistEmptyProcessor.java

@@ -53,7 +53,9 @@ public String identifier() {
 
             @Override
             public PersistProcessor<Boolean> create(
-                    LookupSerializerFactory serializerFactory, @Nullable RowType fileSchema) {
+                    String fileSerVersion,
+                    LookupSerializerFactory serializerFactory,
+                    @Nullable RowType fileSchema) {
                 return new PersistEmptyProcessor();
             }
         };

paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistPositionProcessor.java

@@ -65,7 +65,9 @@ public String identifier() {
 
             @Override
             public PersistProcessor<FilePosition> create(
-                    LookupSerializerFactory serializerFactory, @Nullable RowType fileSchema) {
+                    String fileSerVersion,
+                    LookupSerializerFactory serializerFactory,
+                    @Nullable RowType fileSchema) {
                 return new PersistPositionProcessor();
             }
         };

paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistProcessor.java

@@ -43,6 +43,8 @@ interface Factory<T> {
         String identifier();
 
         PersistProcessor<T> create(
-                LookupSerializerFactory serializerFactory, @Nullable RowType fileSchema);
+                String fileSerVersion,
+                LookupSerializerFactory serializerFactory,
+                @Nullable RowType fileSchema);
     }
 }

paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistValueAndPosProcessor.java

@@ -85,10 +85,13 @@ public String identifier() {
 
             @Override
             public PersistProcessor<PositionedKeyValue> create(
-                    LookupSerializerFactory serializerFactory, @Nullable RowType fileSchema) {
+                    String fileSerVersion,
+                    LookupSerializerFactory serializerFactory,
+                    @Nullable RowType fileSchema) {
                 return new PersistValueAndPosProcessor(
                         serializerFactory.createSerializer(valueType),
-                        serializerFactory.createDeserializer(valueType, fileSchema));
+                        serializerFactory.createDeserializer(
+                                fileSerVersion, valueType, fileSchema));
             }
         };
     }

paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistValueProcessor.java

@@ -73,10 +73,13 @@ public String identifier() {
 
             @Override
             public PersistProcessor<KeyValue> create(
-                    LookupSerializerFactory serializerFactory, @Nullable RowType fileSchema) {
+                    String fileSerVersion,
+                    LookupSerializerFactory serializerFactory,
+                    @Nullable RowType fileSchema) {
                 return new PersistValueProcessor(
                         serializerFactory.createSerializer(valueType),
-                        serializerFactory.createDeserializer(valueType, fileSchema));
+                        serializerFactory.createDeserializer(
+                                fileSerVersion, valueType, fileSchema));
             }
         };
     }

paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/RemoteFileDownloader.java

@@ -25,5 +25,5 @@
 /** Downloader to try to download remote lookup file to local. */
 public interface RemoteFileDownloader {
 
-    boolean tryToDownload(DataFileMeta dataFile, File localFile);
+    boolean tryToDownload(DataFileMeta dataFile, String remoteSstFile, File localFile);
 }