feat(python): meta share mode for pyfory compatible serialization (#2593)

## Why?

implement meta share mode for pyfory, so the struct can add/delete
fields and have inconsistent schema bettween serialization and
deserialization

## What does this PR do?

<!-- Describe the details of this PR. -->

## Related issues

#2509
#1938
#2160
#2278 

## Does this PR introduce any user-facing change?

<!--
If any user-facing interface changes, please [open an
issue](https://github.com/apache/fory/issues/new/choose) describing the
need to do so and update the document if necessary.

Delete section if not applicable.
-->

- [ ] Does this PR introduce any public API change?
- [ ] Does this PR introduce any binary protocol compatibility change?

## Benchmark

<!--
When the PR has an impact on performance (if you don't know whether the
PR will have an impact on performance, you can submit the PR first, and
if it will have impact on performance, the code reviewer will explain
it), be sure to attach a benchmark data here.

Delete section if not applicable.
-->

Author: chaokunyang

ci/format.sh

@@ -125,11 +125,11 @@ format_files() {
 
 format_all_scripts() {
     echo "$(date)" "Ruff format...."
-    git ls-files -- '*.py' '*.pyx' '*.pxd' '*.pxi' "${GIT_LS_EXCLUDES[@]}" | xargs -P 10 \
+    git ls-files -- '*.py' "${GIT_LS_EXCLUDES[@]}" | xargs -P 10 \
       ruff format
 
     echo "$(date)" "Ruff check...."
-    git ls-files -- '*.py' '*.pyx' '*.pxd' '*.pxi' "${GIT_LS_EXCLUDES[@]}" | xargs \
+    git ls-files -- '*.py' "${GIT_LS_EXCLUDES[@]}" | xargs \
       ruff check --fix
 }
 
@@ -193,10 +193,10 @@ format_changed() {
     # exist on both branches.
     MERGEBASE="$(git merge-base origin/main HEAD)"
 
-    if ! git diff --diff-filter=ACRM --quiet --exit-code "$MERGEBASE" -- '*.py' '*.pyx' '*.pxd' '*.pxi' &>/dev/null; then
-        git diff --name-only --diff-filter=ACRM "$MERGEBASE" -- '*.py' '*.pyx' '*.pxd' '*.pxi' | xargs -P 5 \
+    if ! git diff --diff-filter=ACRM --quiet --exit-code "$MERGEBASE" -- '*.py' &>/dev/null; then
+        git diff --name-only --diff-filter=ACRM "$MERGEBASE" -- '*.py' | xargs -P 5 \
             ruff format
-        git diff --name-only --diff-filter=ACRM "$MERGEBASE" -- '*.py' '*.pyx' '*.pxd' '*.pxi' | xargs -P 5 \
+        git diff --name-only --diff-filter=ACRM "$MERGEBASE" -- '*.py' | xargs -P 5 \
             ruff check --fix
     fi
 

python/pyfory/_fory.py

@@ -98,6 +98,7 @@ class Fory:
     __slots__ = (
         "language",
         "is_py",
+        "compatible",
         "ref_tracking",
         "ref_resolver",
         "type_resolver",
@@ -113,13 +114,13 @@ class Fory:
         "_unsupported_objects",
         "_peer_language",
     )
-    serialization_context: "SerializationContext"
 
     def __init__(
         self,
         language=Language.PYTHON,
         ref_tracking: bool = False,
         require_type_registration: bool = True,
+        compatible: bool = False,
     ):
         """
         :param require_type_registration:
@@ -130,10 +131,14 @@ def __init__(
           Do not disable type registration if you can't ensure your environment are
           *indeed secure*. We are not responsible for security risks if
           you disable this option.
+        :param compatible:
+         Whether to enable compatible mode for cross-language serialization.
+         When enabled, type forward/backward compatibility for struct fields will be enabled.
         """
         self.language = language
         self.is_py = language == Language.PYTHON
         self.require_type_registration = _ENABLE_TYPE_REGISTRATION_FORCIBLY or require_type_registration
+        self.compatible = compatible
         self.ref_tracking = ref_tracking
         if self.ref_tracking:
             self.ref_resolver = MapRefResolver()
@@ -143,9 +148,11 @@ def __init__(
         from pyfory._registry import TypeResolver
 
         self.metastring_resolver = MetaStringResolver()
-        self.type_resolver = TypeResolver(self)
+        self.type_resolver = TypeResolver(self, meta_share=compatible)
         self.type_resolver.initialize()
-        self.serialization_context = SerializationContext()
+        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(
@@ -255,10 +262,26 @@ def _serialize(
             set_bit(buffer, mask_index, 3)
         else:
             clear_bit(buffer, mask_index, 3)
+        # Reserve space for type definitions offset, similar to Java implementation
+        type_defs_offset_pos = None
+        if self.serialization_context.scoped_meta_share_enabled:
+            type_defs_offset_pos = buffer.writer_index
+            buffer.write_int32(-1)  # Reserve 4 bytes for type definitions offset
+
         if self.language == Language.PYTHON:
             self.serialize_ref(buffer, obj)
         else:
             self.xserialize_ref(buffer, obj)
+
+        # Write type definitions at the end, similar to Java implementation
+        if self.serialization_context.scoped_meta_share_enabled:
+            meta_context = self.serialization_context.meta_context
+            if meta_context is not None and len(meta_context.get_writing_type_defs()) > 0:
+                # Update the offset to point to current position
+                current_pos = buffer.writer_index
+                buffer.put_int32(type_defs_offset_pos, current_pos - type_defs_offset_pos - 4)
+                self.type_resolver.write_type_defs(buffer)
+
         self.reset_write()
         if buffer is not self.buffer:
             return buffer
@@ -369,6 +392,20 @@ def _deserialize(
             self._buffers = iter(buffers)
         else:
             assert buffers is None, "buffers should be null when the serialized stream is produced with buffer_callback null."
+
+        # Read type definitions at the start, similar to Java implementation
+        if self.serialization_context.scoped_meta_share_enabled:
+            relative_type_defs_offset = buffer.read_int32()
+            if relative_type_defs_offset != -1:
+                # Save current reader position
+                current_reader_index = buffer.reader_index
+                # Jump to type definitions
+                buffer.reader_index = current_reader_index + relative_type_defs_offset
+                # Read type definitions
+                self.type_resolver.read_type_defs(buffer)
+                # Jump back to continue with object deserialization
+                buffer.reader_index = current_reader_index
+
         if is_target_x_lang:
             obj = self.xdeserialize_ref(buffer)
         else:
@@ -470,7 +507,7 @@ def read_ref_pyobject(self, buffer):
     def reset_write(self):
         self.ref_resolver.reset_write()
         self.type_resolver.reset_write()
-        self.serialization_context.reset()
+        self.serialization_context.reset_write()
         self.metastring_resolver.reset_write()
         self.pickler.clear_memo()
         self._buffer_callback = None
@@ -479,7 +516,7 @@ def reset_write(self):
     def reset_read(self):
         self.ref_resolver.reset_read()
         self.type_resolver.reset_read()
-        self.serialization_context.reset()
+        self.serialization_context.reset_read()
         self.metastring_resolver.reset_write()
         self.unpickler = None
         self._buffers = None
@@ -490,36 +527,6 @@ def reset(self):
         self.reset_read()
 
 
-class SerializationContext:
-    """
-    A context is used to add some context-related information, so that the
-    serializers can setup relation between serializing different objects.
-    The context will be reset after finished serializing/deserializing the
-    object tree.
-    """
-
-    __slots__ = ("objects",)
-
-    def __init__(self):
-        self.objects = dict()
-
-    def add(self, key, obj):
-        self.objects[id(key)] = obj
-
-    def __contains__(self, key):
-        return id(key) in self.objects
-
-    def __getitem__(self, key):
-        return self.objects[id(key)]
-
-    def get(self, key):
-        return self.objects.get(id(key))
-
-    def reset(self):
-        if len(self.objects) > 0:
-            self.objects.clear()
-
-
 _ENABLE_TYPE_REGISTRATION_FORCIBLY = os.getenv("ENABLE_TYPE_REGISTRATION_FORCIBLY", "0") in {
     "1",
     "true",