id to version

JingsongLi · JingsongLi · commit be98f6fa88cc · 2025-11-26T12:08:38.000+08:00
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupFile.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupFile.java
@@ -50,7 +50,7 @@ public class LookupFile {
     private final File localFile;
     private final int level;
     private final long schemaId;
-    private final String serializerId;
+    private final String serVersion;
     private final LookupStoreReader reader;
     private final Runnable callback;
 
@@ -62,13 +62,13 @@ public LookupFile(
             File localFile,
             int level,
             long schemaId,
-            String serializerId,
+            String serVersion,
             LookupStoreReader reader,
             Runnable callback) {
         this.localFile = localFile;
         this.level = level;
         this.schemaId = schemaId;
-        this.serializerId = serializerId;
+        this.serVersion = serVersion;
         this.reader = reader;
         this.callback = callback;
     }
@@ -81,8 +81,8 @@ public long schemaId() {
         return schemaId;
     }
 
-    public String serializerId() {
-        return serializerId;
+    public String serVersion() {
+        return serVersion;
     }
 
     @Nullable
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java
@@ -70,7 +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 Map<Pair<Long, String>, PersistProcessor<T>> schemaIdAndSerIdToProcessors;
+    private final Map<Pair<Long, String>, PersistProcessor<T>> schemaIdAndSerVersionToProcessors;
 
     @Nullable private RemoteFileDownloader remoteFileDownloader;
 
@@ -100,7 +100,7 @@ public LookupLevels(
         this.bfGenerator = bfGenerator;
         this.lookupFileCache = lookupFileCache;
         this.ownCachedFiles = new HashSet<>();
-        this.schemaIdAndSerIdToProcessors = new ConcurrentHashMap<>();
+        this.schemaIdAndSerVersionToProcessors = new ConcurrentHashMap<>();
         levels.addDropFileCallback(this);
     }
 
@@ -165,17 +165,17 @@ private T lookup(InternalRow key, DataFileMeta file) throws IOException {
             return null;
         }
 
-        return getOrCreateProcessor(lookupFile.schemaId(), lookupFile.serializerId())
+        return getOrCreateProcessor(lookupFile.schemaId(), lookupFile.serVersion())
                 .readFromDisk(key, lookupFile.level(), valueBytes, file.fileName());
     }
 
-    private PersistProcessor<T> getOrCreateProcessor(long schemaId, String serializerId) {
-        return schemaIdAndSerIdToProcessors.computeIfAbsent(
-                Pair.of(schemaId, serializerId),
+    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(serializerId, serializerFactory, fileSchema);
+                    return processorFactory.create(serVersion, serializerFactory, fileSchema);
                 });
     }
 
@@ -186,12 +186,12 @@ public LookupFile createLookupFile(DataFileMeta file) throws IOException {
         }
 
         long schemaId = this.currentSchemaId;
-        String fileSerId = serializerFactory.identifier();
-        Optional<String> downloadSerId = tryToDownloadRemoteSst(file, localFile);
-        if (downloadSerId.isPresent()) {
+        String fileSerVersion = serializerFactory.version();
+        Optional<String> downloadSerVersion = tryToDownloadRemoteSst(file, localFile);
+        if (downloadSerVersion.isPresent()) {
             // use schema id from remote file
             schemaId = file.schemaId();
-            fileSerId = downloadSerId.get();
+            fileSerVersion = downloadSerVersion.get();
         } else {
             createSstFileFromDataFile(file, localFile);
         }
@@ -201,7 +201,7 @@ public LookupFile createLookupFile(DataFileMeta file) throws IOException {
                 localFile,
                 file.level(),
                 schemaId,
-                fileSerId,
+                fileSerVersion,
                 lookupStoreFactory.createReader(localFile),
                 () -> ownCachedFiles.remove(file.fileName()));
     }
@@ -219,7 +219,7 @@ private Optional<String> tryToDownloadRemoteSst(DataFileMeta file, File localFil
 
         // validate schema matched, no exception here
         try {
-            getOrCreateProcessor(file.schemaId(), remoteSst.serializerId);
+            getOrCreateProcessor(file.schemaId(), remoteSst.serVersion);
         } catch (UnsupportedOperationException e) {
             return Optional.empty();
         }
@@ -229,7 +229,7 @@ private Optional<String> tryToDownloadRemoteSst(DataFileMeta file, File localFil
             return Optional.empty();
         }
 
-        return Optional.of(remoteSst.serializerId);
+        return Optional.of(remoteSst.serVersion);
     }
 
     public void addLocalFile(DataFileMeta file, LookupFile lookupFile) {
@@ -242,7 +242,7 @@ private void createSstFileFromDataFile(DataFileMeta file, File localFile) throws
                                 localFile, bfGenerator.apply(file.rowCount()));
                 RecordReader<KeyValue> reader = fileReaderFactory.apply(file)) {
             PersistProcessor<T> processor =
-                    getOrCreateProcessor(currentSchemaId, serializerFactory.identifier());
+                    getOrCreateProcessor(currentSchemaId, serializerFactory.version());
             KeyValue kv;
             if (processor.withPosition()) {
                 FileRecordIterator<KeyValue> batch;
@@ -291,8 +291,8 @@ public Optional<RemoteSstFile> remoteSst(DataFileMeta file) {
             return Optional.empty();
         }
 
-        String serializerId = split[split.length - 2];
-        return Optional.of(new RemoteSstFile(sstFileName, serializerId));
+        String serVersion = split[split.length - 2];
+        return Optional.of(new RemoteSstFile(sstFileName, serVersion));
     }
 
     public String newRemoteSst(DataFileMeta file, long length) {
@@ -302,7 +302,7 @@ public String newRemoteSst(DataFileMeta file, long length) {
                 + "."
                 + processorFactory.identifier()
                 + "."
-                + serializerFactory.identifier()
+                + serializerFactory.version()
                 + REMOTE_LOOKUP_FILE_SUFFIX;
     }
 
@@ -314,15 +314,15 @@ public void close() throws IOException {
         }
     }
 
-    /** Remote sst file with serializerId. */
+    /** Remote sst file with serVersion. */
     public static class RemoteSstFile {
 
         private final String sstFileName;
-        private final String serializerId;
+        private final String serVersion;
 
-        private RemoteSstFile(String sstFileName, String serializerId) {
+        private RemoteSstFile(String sstFileName, String serVersion) {
             this.sstFileName = sstFileName;
-            this.serializerId = serializerId;
+            this.serVersion = serVersion;
         }
     }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/DefaultLookupSerializerFactory.java b/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,8 +42,8 @@ public Function<InternalRow, byte[]> createSerializer(RowType currentSchema) {
 
     @Override
     public Function<byte[], InternalRow> createDeserializer(
-            String fileSerializerId, RowType currentSchema, @Nullable RowType fileSchema) {
-        if (!identifier().equals(fileSerializerId)) {
+            String fileSerVersion, RowType currentSchema, @Nullable RowType fileSchema) {
+        if (!version().equals(fileSerVersion)) {
             throw new UnsupportedOperationException();
         }
         if (fileSchema != null && !fileSchema.equalsIgnoreNullable(currentSchema)) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/LookupSerializerFactory.java b/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(
-            String fileSerializerId, RowType currentSchema, @Nullable RowType fileSchema);
+            String fileSerVersion, RowType currentSchema, @Nullable RowType fileSchema);
 
     Supplier<LookupSerializerFactory> INSTANCE =
             Suppliers.memoize(
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistEmptyProcessor.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistEmptyProcessor.java
@@ -53,7 +53,7 @@ public String identifier() {
 
             @Override
             public PersistProcessor<Boolean> create(
-                    String fileSerializerId,
+                    String fileSerVersion,
                     LookupSerializerFactory serializerFactory,
                     @Nullable RowType fileSchema) {
                 return new PersistEmptyProcessor();
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistPositionProcessor.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistPositionProcessor.java
@@ -65,7 +65,7 @@ public String identifier() {
 
             @Override
             public PersistProcessor<FilePosition> create(
-                    String fileSerializerId,
+                    String fileSerVersion,
                     LookupSerializerFactory serializerFactory,
                     @Nullable RowType fileSchema) {
                 return new PersistPositionProcessor();
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistProcessor.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistProcessor.java
@@ -43,7 +43,7 @@ interface Factory<T> {
         String identifier();
 
         PersistProcessor<T> create(
-                String fileSerializerId,
+                String fileSerVersion,
                 LookupSerializerFactory serializerFactory,
                 @Nullable RowType fileSchema);
     }
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistValueAndPosProcessor.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistValueAndPosProcessor.java
@@ -85,13 +85,13 @@ public String identifier() {
 
             @Override
             public PersistProcessor<PositionedKeyValue> create(
-                    String fileSerializerId,
+                    String fileSerVersion,
                     LookupSerializerFactory serializerFactory,
                     @Nullable RowType fileSchema) {
                 return new PersistValueAndPosProcessor(
                         serializerFactory.createSerializer(valueType),
                         serializerFactory.createDeserializer(
-                                fileSerializerId, valueType, fileSchema));
+                                fileSerVersion, valueType, fileSchema));
             }
         };
     }
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistValueProcessor.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/lookup/PersistValueProcessor.java
@@ -73,13 +73,13 @@ public String identifier() {
 
             @Override
             public PersistProcessor<KeyValue> create(
-                    String fileSerializerId,
+                    String fileSerVersion,
                     LookupSerializerFactory serializerFactory,
                     @Nullable RowType fileSchema) {
                 return new PersistValueProcessor(
                         serializerFactory.createSerializer(valueType),
                         serializerFactory.createDeserializer(
-                                fileSerializerId, valueType, fileSchema));
+                                fileSerVersion, valueType, fileSchema));
             }
         };
     }

Original file line number	Diff line number	Diff line change
`@@ -43,7 +43,7 @@ interface Factory<T> {`
`43`	`43`	`String identifier();`
`44`	`44`
`45`	`45`	`PersistProcessor<T> create(`
`46`		`- String fileSerializerId,`
	`46`	`+ String fileSerVersion,`
`47`	`47`	`LookupSerializerFactory serializerFactory,`
`48`	`48`	`@Nullable RowType fileSchema);`
`49`	`49`	`}`