feat(java/python): align java and python compatible mode serialization

cpp/fory/type/type.h

@@ -119,4 +119,16 @@ inline bool IsNamespacedType(int32_t type_id) {
   }
 }
 
+inline bool IsTypeShareMeta(int32_t type_id) {
+  switch (static_cast<TypeId>(type_id)) {
+  case TypeId::NAMED_ENUM:
+  case TypeId::NAMED_STRUCT:
+  case TypeId::NAMED_EXT:
+  case TypeId::COMPATIBLE_STRUCT:
+  case TypeId::NAMED_COMPATIBLE_STRUCT:
+    return true;
+  default:
+    return false;
+  }
+}
 } // namespace fory

java/fory-core/src/main/java/org/apache/fory/Fory.java

@@ -1703,6 +1703,10 @@ public boolean isCrossLanguage() {
     return crossLanguage;
   }
 
+  public boolean isCompatible() {
+    return config.getCompatibleMode() == CompatibleMode.COMPATIBLE;
+  }
+
   public boolean trackingRef() {
     return refTracking;
   }

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

@@ -1725,16 +1725,11 @@ public void writeClassDefs(MemoryBuffer buffer) {
 
   private void writeClassDefs(
       MemoryBuffer buffer, ObjectArray<ClassDef> writingClassDefs, int size) {
-    int writerIndex = buffer.writerIndex();
     for (int i = 0; i < size; i++) {
-      byte[] encoded = writingClassDefs.get(i).getEncoded();
-      int bytesLen = encoded.length;
-      buffer.ensure(writerIndex + bytesLen);
-      final byte[] targetArray = buffer.getHeapMemory();
-      System.arraycopy(encoded, 0, targetArray, writerIndex, bytesLen);
-      writerIndex += bytesLen;
-    }
-    buffer.writerIndex(writerIndex);
+      buffer.writeBytes(writingClassDefs.get(i).getEncoded());
+      MemoryBuffer memoryBuffer = MemoryBuffer.fromByteArray(writingClassDefs.get(i).getEncoded());
+      ClassDef.readClassDef(fory, memoryBuffer, memoryBuffer.readInt64());
+    }
   }
 
   /**

java/fory-core/src/main/java/org/apache/fory/resolver/XtypeResolver.java

@@ -38,11 +38,13 @@
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -210,7 +212,8 @@ public void register(Class<?> type, String namespace, String typeName) {
     short xtypeId;
     if (serializer != null) {
       if (isStructType(serializer)) {
-        xtypeId = Types.NAMED_STRUCT;
+        xtypeId =
+            (short) (fory.isCompatible() ? Types.NAMED_COMPATIBLE_STRUCT : Types.NAMED_STRUCT);
       } else if (serializer instanceof EnumSerializer) {
         xtypeId = Types.NAMED_ENUM;
       } else {
@@ -220,7 +223,8 @@ public void register(Class<?> type, String namespace, String typeName) {
       if (type.isEnum()) {
         xtypeId = Types.NAMED_ENUM;
       } else {
-        xtypeId = Types.NAMED_STRUCT;
+        xtypeId =
+            (short) (fory.isCompatible() ? Types.NAMED_COMPATIBLE_STRUCT : Types.NAMED_STRUCT);
       }
     }
     register(type, serializer, namespace, typeName, xtypeId);
@@ -530,9 +534,9 @@ private void registerDefaultTypes() {
     registerDefaultTypes(Types.INT64_ARRAY, long[].class);
     registerDefaultTypes(Types.FLOAT32_ARRAY, float[].class);
     registerDefaultTypes(Types.FLOAT64_ARRAY, double[].class);
-    registerDefaultTypes(Types.LIST, ArrayList.class, Object[].class);
-    registerDefaultTypes(Types.SET, HashSet.class, LinkedHashSet.class);
-    registerDefaultTypes(Types.MAP, HashMap.class, LinkedHashMap.class);
+    registerDefaultTypes(Types.LIST, ArrayList.class, Object[].class, List.class, Collection.class);
+    registerDefaultTypes(Types.SET, HashSet.class, LinkedHashSet.class, Set.class);
+    registerDefaultTypes(Types.MAP, HashMap.class, LinkedHashMap.class, Map.class);
     registerDefaultTypes(Types.LOCAL_DATE, LocalDate.class);
   }
 
@@ -542,8 +546,12 @@ private void registerDefaultTypes(int xtypeId, Class<?> defaultType, Class<?>...
     classInfoMap.put(defaultType, classInfo);
     xtypeIdToClassMap.put(xtypeId, classInfo);
     for (Class<?> otherType : otherTypes) {
-      classInfo = newClassInfo(otherType, classResolver.getSerializer(otherType), (short) xtypeId);
-      classInfoMap.put(otherType, classInfo);
+      Serializer<?> serializer =
+          ReflectionUtils.isAbstract(otherType)
+              ? classInfo.serializer
+              : classResolver.getSerializer(otherType);
+      ClassInfo info = newClassInfo(otherType, serializer, (short) xtypeId);
+      classInfoMap.put(otherType, info);
     }
   }
 
@@ -561,7 +569,6 @@ public void writeClassInfo(MemoryBuffer buffer, ClassInfo classInfo) {
     switch (internalTypeId) {
       case Types.NAMED_ENUM:
       case Types.NAMED_STRUCT:
-      case Types.NAMED_COMPATIBLE_STRUCT:
       case Types.NAMED_EXT:
         if (shareMeta) {
           writeSharedClassMeta(buffer, classInfo);
@@ -572,6 +579,11 @@ public void writeClassInfo(MemoryBuffer buffer, ClassInfo classInfo) {
         assert classInfo.typeNameBytes != null;
         metaStringResolver.writeMetaStringBytes(buffer, classInfo.typeNameBytes);
         break;
+      case Types.NAMED_COMPATIBLE_STRUCT:
+      case Types.COMPATIBLE_STRUCT:
+        assert shareMeta : "Meta share must be enabled for compatible mode";
+        writeSharedClassMeta(buffer, classInfo);
+        break;
       default:
         break;
     }
@@ -634,14 +646,17 @@ public ClassInfo readClassInfo(MemoryBuffer buffer) {
     switch (internalTypeId) {
       case Types.NAMED_ENUM:
       case Types.NAMED_STRUCT:
-      case Types.NAMED_COMPATIBLE_STRUCT:
       case Types.NAMED_EXT:
         if (shareMeta) {
           return readSharedClassMeta(buffer);
         }
         MetaStringBytes packageBytes = metaStringResolver.readMetaStringBytes(buffer);
         MetaStringBytes simpleClassNameBytes = metaStringResolver.readMetaStringBytes(buffer);
         return loadBytesToClassInfo(internalTypeId, packageBytes, simpleClassNameBytes);
+      case Types.NAMED_COMPATIBLE_STRUCT:
+      case Types.COMPATIBLE_STRUCT:
+        assert shareMeta : "Meta share must be enabled for compatible mode";
+        return readSharedClassMeta(buffer);
       case Types.LIST:
         return getListClassInfo();
       case Types.TIMESTAMP:

java/fory-core/src/test/java/org/apache/fory/CrossLanguageTest.java

@@ -483,19 +483,21 @@ public void testStructHash() throws Exception {
     roundBytes("test_struct_hash", buffer.getBytes(0, 4));
   }
 
-  @Test
-  public void testSerializeSimpleStruct() throws Exception {
+  @Test(dataProvider = "compatible")
+  public void testSerializeSimpleStruct(boolean compatible) throws Exception {
     Fory fory =
         Fory.builder()
             .withLanguage(Language.XLANG)
+            .withCompatibleMode(
+                compatible ? CompatibleMode.COMPATIBLE : CompatibleMode.SCHEMA_CONSISTENT)
             .withRefTracking(true)
             .requireClassRegistration(false)
             .build();
     fory.register(ComplexObject2.class, "test.ComplexObject2");
     ComplexObject2 obj2 = new ComplexObject2();
     obj2.f1 = true;
     obj2.f2 = new HashMap<>(ImmutableMap.of((byte) -1, 2));
-    structRoundBack(fory, obj2, "test_serialize_simple_struct");
+    structRoundBack(fory, obj2, "test_serialize_simple_struct" + (compatible ? "_compatible" : ""));
   }
 
   @Test
@@ -530,10 +532,13 @@ public void testRegisterByIdMetaShare() throws Exception {
     Assert.assertEquals(fory.deserialize(serialized), obj);
   }
 
-  public void testSerializeComplexStruct() throws Exception {
+  @Test(dataProvider = "compatible")
+  public void testSerializeComplexStruct(boolean compatible) throws Exception {
     Fory fory =
         Fory.builder()
             .withLanguage(Language.XLANG)
+            .withCompatibleMode(
+                compatible ? CompatibleMode.COMPATIBLE : CompatibleMode.SCHEMA_CONSISTENT)
             .withRefTracking(true)
             .requireClassRegistration(false)
             .build();
@@ -556,7 +561,7 @@ public void testSerializeComplexStruct() throws Exception {
     obj.f11 = new short[] {(short) 1, (short) 2};
     obj.f12 = ImmutableList.of((short) -1, (short) 4);
 
-    structRoundBack(fory, obj, "test_serialize_complex_struct");
+    structRoundBack(fory, obj, "test_serialize_complex_struct" + (compatible ? "_compatible" : ""));
   }
 
   private void structRoundBack(Fory fory, Object obj, String testName) throws IOException {

java/fory-core/src/test/java/org/apache/fory/ForyTestBase.java

@@ -131,6 +131,11 @@ public static Object[][] enableCodegen() {
     return new Object[][] {{false}, {true}};
   }
 
+  @DataProvider
+  public static Object[][] compatible() {
+    return new Object[][] {{false}, {true}};
+  }
+
   @DataProvider
   public static Object[][] compressNumber() {
     return new Object[][] {{false}, {true}};

python/pyfory/_fory.py

@@ -144,15 +144,14 @@ def __init__(
             self.ref_resolver = MapRefResolver()
         else:
             self.ref_resolver = NoRefResolver()
-        from pyfory._serialization import MetaStringResolver
+        from pyfory._serialization import MetaStringResolver, SerializationContext
         from pyfory._registry import TypeResolver
 
         self.metastring_resolver = MetaStringResolver()
         self.type_resolver = TypeResolver(self, meta_share=compatible)
+        self.serialization_context = SerializationContext(fory=self, scoped_meta_share_enabled=compatible)
         self.type_resolver.initialize()
-        from pyfory._serialization import SerializationContext
 
-        self.serialization_context = SerializationContext(scoped_meta_share_enabled=compatible)
         self.buffer = Buffer.allocate(32)
         if not require_type_registration:
             warnings.warn(

python/pyfory/_registry.py

@@ -169,6 +169,7 @@ class TypeResolver:
         "_type_id_to_typeinfo",
         "_meta_shared_typeinfo",
         "meta_share",
+        "serialization_context",
     )
 
     def __init__(self, fory, meta_share=False):
@@ -202,6 +203,7 @@ def initialize(self):
         self._initialize_xlang()
         if self.fory.language == Language.PYTHON:
             self._initialize_py()
+        self.serialization_context = self.fory.serialization_context
 
     def _initialize_py(self):
         register = functools.partial(self._register_type, internal=True)
@@ -369,7 +371,10 @@ def _register_xtype(
                 type_id = TypeId.NAMED_EXT if type_id is None else ((type_id << 8) + TypeId.EXT)
             else:
                 serializer = None
-                type_id = TypeId.NAMED_STRUCT if type_id is None else ((type_id << 8) + TypeId.STRUCT)
+                if self.meta_share:
+                    type_id = TypeId.NAMED_COMPATIBLE_STRUCT if type_id is None else ((type_id << 8) + TypeId.COMPATIBLE_STRUCT)
+                else:
+                    type_id = TypeId.NAMED_STRUCT if type_id is None else ((type_id << 8) + TypeId.STRUCT)
         elif not internal:
             type_id = TypeId.NAMED_EXT if type_id is None else ((type_id << 8) + TypeId.EXT)
         return self.__register_type(
@@ -607,14 +612,12 @@ def _load_metabytes_to_typeinfo(self, ns_metabytes, type_metabytes):
     def write_typeinfo(self, buffer, typeinfo):
         if typeinfo.dynamic_type:
             return
-        type_id = typeinfo.type_id
-        internal_type_id = type_id & 0xFF
-
         # Check if meta share is enabled first
         if self.meta_share:
             self.write_shared_type_meta(buffer, typeinfo)
             return
-
+        type_id = typeinfo.type_id
+        internal_type_id = type_id & 0xFF
         buffer.write_varuint32(type_id)
         if TypeId.is_namespaced_type(internal_type_id):
             self.metastring_resolver.write_meta_string_bytes(buffer, typeinfo.namespace_bytes)
@@ -659,17 +662,14 @@ def get_meta_compressor(self):
 
     def write_shared_type_meta(self, buffer, typeinfo):
         """Write shared type meta information."""
-        assert typeinfo.type_def is not None, "Type info must be set when meta share is enabled"
         meta_context = self.fory.serialization_context.meta_context
-        meta_context.write_typeinfo(buffer, typeinfo)
+        meta_context.write_shared_typeinfo(buffer, typeinfo)
 
     def read_shared_type_meta(self, buffer):
         """Read shared type meta information."""
-        meta_context = self.fory.serialization_context.meta_context
+        meta_context = self.serialization_context.meta_context
         assert meta_context is not None, "Meta context must be set when meta share is enabled"
-        type_id = buffer.read_varuint32()
-        typeinfo = meta_context.get_read_type_info(type_id)
-        assert typeinfo is not None, f"Type info not found for ID {type_id}"
+        typeinfo = meta_context.read_shared_typeinfo(buffer)
         return typeinfo
 
     def write_type_defs(self, buffer):
@@ -704,7 +704,7 @@ def read_type_defs(self, buffer):
                 type_info = self._build_type_info_from_typedef(type_def)
                 # Cache the tuple for future use
                 self._meta_shared_typeinfo[header] = type_info
-            meta_context.add_read_type_info(type_info)
+            meta_context.add_read_typeinfo(type_info)
 
     def _build_type_info_from_typedef(self, type_def):
         """Build TypeInfo from TypeDef using TypeDef's create_serializer method."""