support deserialize one class into another in meta share mode

java/fury-core/src/main/java/org/apache/fury/Fury.java

@@ -1122,7 +1122,7 @@ public <T> T deserializeJavaObject(MemoryBuffer buffer, Class<T> cls) {
       if (nextReadRefId >= NOT_NULL_VALUE_FLAG) {
         ClassInfo classInfo;
         if (shareMeta) {
-          classInfo = classResolver.readClassInfo(buffer, cls);
+          classInfo = classResolver.readClassInfoWithMetaShare(buffer, cls);
         } else {
           classInfo = classResolver.getClassInfo(cls);
         }

java/fury-core/src/main/java/org/apache/fury/meta/ClassDef.java

@@ -37,6 +37,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.SortedMap;
+import java.util.stream.Collectors;
 import org.apache.fury.Fury;
 import org.apache.fury.builder.MetaSharedCodecBuilder;
 import org.apache.fury.collection.Tuple2;
@@ -48,7 +49,6 @@
 import org.apache.fury.memory.Platform;
 import org.apache.fury.reflect.ReflectionUtils;
 import org.apache.fury.reflect.TypeRef;
-import org.apache.fury.resolver.ClassInfo;
 import org.apache.fury.resolver.ClassResolver;
 import org.apache.fury.serializer.CompatibleSerializer;
 import org.apache.fury.serializer.NonexistentClass;
@@ -789,11 +789,20 @@ public static ClassDef buildClassDef(
     return ClassDefEncoder.buildClassDef(classResolver, type, fields, isObjectType);
   }
 
-  public static ClassDef replaceRootClassTo(
-      ClassResolver classResolver,
-      ClassInfo targetCls
-  ) {
+  public ClassDef replaceRootClassTo(ClassResolver classResolver, Class<?> targetCls) {
+    String name = targetCls.getName();
+    List<FieldInfo> fieldInfos =
+        fieldsInfo.stream()
+            .map(
+                fieldInfo -> {
+                  if (fieldInfo.definedClass.equals(classSpec.entireClassName)) {
+                    return new FieldInfo(name, fieldInfo.fieldName, fieldInfo.fieldType);
+                  } else {
+                    return fieldInfo;
+                  }
+                })
+            .collect(Collectors.toList());
     return ClassDefEncoder.buildClassDefWithFieldInfos(
-        classResolver, targetCls.getCls(), targetCls.classDef.getFieldsInfo(), targetCls.classDef.isObjectType);
+        classResolver, targetCls, fieldInfos, isObjectType);
   }
 }

java/fury-core/src/main/java/org/apache/fury/meta/ClassDefEncoder.java

@@ -105,11 +105,15 @@ static List<FieldInfo> buildFieldsInfo(ClassResolver resolver, List<Field> field
   /** Build class definition from fields of class. */
   static ClassDef buildClassDef(
       ClassResolver classResolver, Class<?> type, List<Field> fields, boolean isObjectType) {
-    return buildClassDefWithFieldInfos(classResolver, type, buildFieldsInfo(classResolver, fields), isObjectType);
+    return buildClassDefWithFieldInfos(
+        classResolver, type, buildFieldsInfo(classResolver, fields), isObjectType);
   }
 
   static ClassDef buildClassDefWithFieldInfos(
-      ClassResolver classResolver, Class<?> type, List<ClassDef.FieldInfo> fieldInfos, boolean isObjectType) {
+      ClassResolver classResolver,
+      Class<?> type,
+      List<ClassDef.FieldInfo> fieldInfos,
+      boolean isObjectType) {
     Map<String, List<FieldInfo>> classLayers = getClassFields(type, fieldInfos);
     fieldInfos = new ArrayList<>(fieldInfos.size());
     classLayers.values().forEach(fieldInfos::addAll);

java/fury-core/src/main/java/org/apache/fury/meta/ClassSpec.java

@@ -22,7 +22,7 @@
 import org.apache.fury.type.TypeUtils;
 
 public class ClassSpec {
-  public String entireClassName;
+  public final String entireClassName;
 
   /** Whether current class is enum of component is enum if current class is array. */
   public final boolean isEnum;

java/fury-core/src/main/java/org/apache/fury/resolver/ClassInfo.java

@@ -48,7 +48,7 @@ public class ClassInfo {
   // use primitive to avoid boxing
   // class id must be less than Integer.MAX_VALUE/2 since we use bit 0 as class id flag.
   short classId;
-  public ClassDef classDef;
+  ClassDef classDef;
   boolean needToWriteClassDef;
 
   ClassInfo(

java/fury-core/src/main/java/org/apache/fury/resolver/ClassResolver.java

@@ -247,6 +247,8 @@ private static class ExtRegistry {
     private final Map<Class<?>, FieldResolver> fieldResolverMap = new HashMap<>();
     private final LongMap<Tuple2<ClassDef, ClassInfo>> classIdToDef = new LongMap<>();
     private final Map<Class<?>, ClassDef> currentLayerClassDef = new HashMap<>();
+    // Tuple2<Class, Class>: Tuple2<From Class, To Class>
+    private final Map<Tuple2<Class<?>, Class<?>>, ClassInfo> transformedClassInfo = new HashMap<>();
     // TODO(chaokunyang) Better to  use soft reference, see ObjectStreamClass.
     private final ConcurrentHashMap<Tuple2<Class<?>, Boolean>, SortedMap<Field, Descriptor>>
         descriptorsCache = new ConcurrentHashMap<>();
@@ -1411,6 +1413,27 @@ private ClassInfo readClassInfoWithMetaShare(MetaContext metaContext, int index)
     return classInfo;
   }
 
+  public ClassInfo readClassInfoWithMetaShare(MemoryBuffer buffer, Class<?> targetClass) {
+    assert metaContextShareEnabled;
+    ClassInfo classInfo =
+        readClassInfoWithMetaShare(buffer, fury.getSerializationContext().getMetaContext());
+    Class<?> readClass = classInfo.getCls();
+    // replace target class if needed
+    if (targetClass != readClass) {
+      Tuple2<Class<?>, Class<?>> key = Tuple2.of(readClass, targetClass);
+      ClassInfo newClassInfo = extRegistry.transformedClassInfo.get(key);
+      if (newClassInfo == null) {
+        // similar to create serializer for `NonexistentMetaShared`
+        newClassInfo =
+            getMetaSharedClassInfo(
+                classInfo.classDef.replaceRootClassTo(this, targetClass), targetClass);
+        extRegistry.transformedClassInfo.put(key, newClassInfo);
+      }
+      return newClassInfo;
+    }
+    return classInfo;
+  }
+
   private ClassInfo buildMetaSharedClassInfo(
       Tuple2<ClassDef, ClassInfo> classDefTuple, ClassDef classDef) {
     ClassInfo classInfo;
@@ -1439,6 +1462,7 @@ private ClassInfo getMetaSharedClassInfo(ClassDef classDef, Class<?> clz) {
     Short classId = extRegistry.registeredClassIdMap.get(cls);
     ClassInfo classInfo =
         new ClassInfo(this, cls, null, classId == null ? NO_CLASS_ID : classId, NOT_SUPPORT_XLANG);
+    classInfo.classDef = classDef;
     if (NonexistentClass.class.isAssignableFrom(TypeUtils.getComponentIfArray(cls))) {
       if (cls == NonexistentMetaShared.class) {
         classInfo.setSerializer(this, new NonexistentClassSerializer(fury, classDef));
@@ -1674,33 +1698,6 @@ public ClassInfo readClassInfo(MemoryBuffer buffer) {
     return classInfo;
   }
 
-  public ClassInfo readClassInfo(MemoryBuffer buffer, Class<?> targetClass) {
-    if (metaContextShareEnabled) {
-      ClassInfo classInfo = readClassInfoWithMetaShare(buffer, fury.getSerializationContext().getMetaContext());
-
-      // replace target class if needed
-      if (!targetClass.getName().equals(classInfo.getCls().getName())) {
-        ClassInfo targetClassInfo = getClassInfo(targetClass);
-        buildClassDef(targetClassInfo);
-        classInfo.classDef = ClassDef.replaceRootClassTo(this, targetClassInfo);
-        classInfo.serializer = createSerializer(targetClass);
-      }
-
-      return classInfo;
-    }
-
-    int header = buffer.readVarUint32Small14();
-    ClassInfo classInfo;
-    if ((header & 0b1) != 0) {
-      classInfo = readClassInfoFromBytes(buffer, classInfoCache, header);
-      classInfoCache = classInfo;
-    } else {
-      classInfo = getOrUpdateClassInfo((short) (header >> 1));
-    }
-    currentReadClass = classInfo.cls;
-    return classInfo;
-  }
-
   /**
    * Read class info from java data <code>buffer</code>. `classInfoCache` is used as a cache to
    * reduce map lookup to load class from binary.

java/fury-core/src/test/java/org/apache/fury/serializer/compatible/DifferentPOJOCompatibleSerializerTest.java

@@ -76,8 +76,7 @@ void testTargetHasMoreFieldComparedToSourceClass() throws InterruptedException {
     byte[] serialized = getFury().serializeJavaObject(classMissingField);
 
     ClassCompleteField classCompleteField =
-        getFury()
-            .deserializeJavaObject(serialized, ClassCompleteField.class);
+        getFury().deserializeJavaObject(serialized, ClassCompleteField.class);
 
     assertEq(classCompleteField, classMissingField);
   }