feat(go): Implement compatible mode with metashare mode (#2607)

## Why?

Type-forward and backward-compatible serialization are crucial for
online services where different services update their data schemas and
deploy at different times. This ensures that a service running an older
schema can still communicate with a service using a newer one, and vice
versa.This PR is part of the work of it. The previous PR is #2554 , and
following pr are still work in process.

## What does this PR do?

This PR implements a metashare mode for the fory-go.

Specifically, it:

Provides a user option to enable compatible mode.

If compatible mode is enabled, the serialization process not only
serializes the data but also its detailed type information into a binary
format.

The reading peer, if compatible mode is enabled, will first attempt to
read this type information before deserializing the data. This allows it
to handle data serialized with a different schema.

This implementation is heavily inspired by the existing Java and Python
implementations, ensuring consistency across framework.

All progress and todos:
- [x] metadata encoding and decoding
- [x] Integrate into the main flow to enable metadata sharing
- [  ] Add support for collection types (map, slice, set, etc.)
- [  ] Add support for compressed type definitions
- [  ] Test with python

## Related issues

#2192

## 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.
-->

- [x] Does this PR introduce any public API change? Yes.
- [x] Does this PR introduce any binary protocol compatibility change?
No.

## 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: junjiexh

.gitignore

@@ -40,3 +40,4 @@ MODULE.bazel
 MODULE.bazel.lock
 .DS_Store
 **/.DS_Store
+.vscode/

go/README.md

@@ -146,6 +146,38 @@ fory --force -file <your file>
 - Does generated code work across Go versions? Yes, it’s plain Go code; keep your toolchain consistent in CI.
 - Can I mix generated and non-generated structs? Yes, adoption is incremental and per file.
 
+## Configuration Options
+
+Fory Go supports several configuration options through the functional options pattern:
+
+### Compatible Mode (Metashare)
+
+Compatible mode enables meta information sharing, which allows for schema evolution:
+
+```go
+// Enable compatible mode with metashare
+fory := NewForyWithOptions(WithCompatible(true))
+```
+
+### Reference Tracking
+
+Enable reference tracking:
+
+```go
+fory := NewForyWithOptions(WithRefTracking(true))
+```
+
+### Combined Options
+
+You can combine multiple options:
+
+```go
+fory := NewForyWithOptions(
+    WithCompatible(true),
+    WithRefTracking(true),
+)
+```
+
 ## How to test
 
 ```bash

go/fory/context.go

@@ -0,0 +1,72 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package fory
+
+import "reflect"
+
+// MetaContext used to share data across multiple serialization calls
+type MetaContext struct {
+	// typeMap make sure each type just fully serialize once, the following serialization will use the index
+	typeMap map[reflect.Type]uint32
+	// record typeDefs need to be serialized during one serialization
+	writingTypeDefs []*TypeDef
+	// read from peer
+	readTypeInfos []TypeInfo
+	// scopedMetaShareEnabled controls whether meta sharing is scoped to single serialization
+	scopedMetaShareEnabled bool
+}
+
+// NewMetaContext creates a new MetaContext
+func NewMetaContext(scopedMetaShareEnabled bool) *MetaContext {
+	return &MetaContext{
+		typeMap:                make(map[reflect.Type]uint32),
+		scopedMetaShareEnabled: scopedMetaShareEnabled,
+	}
+}
+
+// resetRead resets the read-related state of the MetaContext
+func (mc *MetaContext) resetRead() {
+	if mc.scopedMetaShareEnabled {
+		mc.readTypeInfos = mc.readTypeInfos[:0] // Reset slice but keep capacity
+	} else {
+		mc.readTypeInfos = nil
+	}
+}
+
+// resetWrite resets the write-related state of the MetaContext
+func (mc *MetaContext) resetWrite() {
+	if mc.scopedMetaShareEnabled {
+		for k := range mc.typeMap {
+			delete(mc.typeMap, k)
+		}
+		mc.writingTypeDefs = mc.writingTypeDefs[:0] // Reset slice but keep capacity
+	} else {
+		mc.typeMap = nil
+		mc.writingTypeDefs = nil
+	}
+}
+
+// SetScopedMetaShareEnabled sets the scoped meta share mode
+func (mc *MetaContext) SetScopedMetaShareEnabled(enabled bool) {
+	mc.scopedMetaShareEnabled = enabled
+}
+
+// IsScopedMetaShareEnabled returns whether scoped meta sharing is enabled
+func (mc *MetaContext) IsScopedMetaShareEnabled() bool {
+	return mc.scopedMetaShareEnabled
+}

go/fory/fory.go

@@ -24,15 +24,72 @@ import (
 	"sync"
 )
 
-func NewFory(referenceTracking bool) *Fory {
+// Option represents a configuration option for Fory instances.
+// This follows the functional options pattern, allowing flexible configuration
+// by passing variadic option functions to constructors like NewForyWithOptions.
+type Option func(*Fory)
+
+// WithCompatible sets the compatible mode for the Fory instance.
+// When compatible=true, scoped meta sharing is automatically enabled.
+func WithCompatible(compatible bool) Option {
+	return func(f *Fory) {
+		f.compatible = compatible
+		if compatible {
+			f.metaContext = NewMetaContext(true) // Enable scoped meta sharing
+		} else {
+			f.metaContext = nil
+		}
+	}
+}
+
+// WithRefTracking sets the reference tracking mode for the Fory instance
+func WithRefTracking(refTracking bool) Option {
+	return func(f *Fory) {
+		f.refTracking = refTracking
+	}
+}
+
+// WithScopedMetaShare sets the scoped meta share mode for the Fory instance.
+// Note: Compatible mode automatically enables scoped meta sharing.
+// This option is mainly used for fine-grained control when compatible mode is already enabled.
+func WithScopedMetaShare(enabled bool) Option {
+	return func(f *Fory) {
+		if f.metaContext == nil {
+			f.metaContext = NewMetaContext(enabled)
+		} else {
+			f.metaContext.SetScopedMetaShareEnabled(enabled)
+		}
+	}
+}
+
+func NewFory(refTracking bool) *Fory {
+	return NewForyWithOptions(WithRefTracking(refTracking))
+}
+
+// NewForyWithOptions creates a Fory instance with configurable options
+func NewForyWithOptions(options ...Option) *Fory {
 	fory := &Fory{
-		refResolver:       newRefResolver(referenceTracking),
-		referenceTracking: referenceTracking,
-		language:          XLANG,
-		buffer:            NewByteBuffer(nil),
+		refResolver: nil,
+		refTracking: false,
+		language:    XLANG,
+		buffer:      NewByteBuffer(nil),
+		compatible:  false,
 	}
-	// Create a new type resolver for this instance
+
+	// Apply options
+	for _, option := range options {
+		option(fory)
+	}
+
+	// Create a new type resolver for this instance but copy generated serializers from global resolver
 	fory.typeResolver = newTypeResolver(fory)
+
+	// Initialize meta context if compatible mode is enabled
+	if fory.compatible {
+		fory.metaContext = NewMetaContext(true)
+	}
+
+	fory.refResolver = newRefResolver(fory.refTracking)
 	return fory
 }
 
@@ -105,14 +162,16 @@ const (
 const MAGIC_NUMBER int16 = 0x62D4
 
 type Fory struct {
-	typeResolver      *typeResolver
-	refResolver       *RefResolver
-	referenceTracking bool
-	language          Language
-	bufferCallback    BufferCallback
-	peerLanguage      Language
-	buffer            *ByteBuffer
-	buffers           []*ByteBuffer
+	typeResolver   *typeResolver
+	refResolver    *RefResolver
+	refTracking    bool
+	language       Language
+	bufferCallback BufferCallback
+	peerLanguage   Language
+	buffer         *ByteBuffer
+	buffers        []*ByteBuffer
+	compatible     bool
+	metaContext    *MetaContext
 }
 
 func (f *Fory) RegisterTagType(tag string, v interface{}) error {
@@ -246,7 +305,18 @@ func (f *Fory) readLength(buffer *ByteBuffer) int {
 }
 
 func (f *Fory) WriteReferencable(buffer *ByteBuffer, value reflect.Value) error {
-	return f.writeReferencableBySerializer(buffer, value, nil)
+	metaOffset := buffer.writerIndex
+	if f.compatible {
+		buffer.WriteInt32(-1)
+	}
+	if err := f.writeReferencableBySerializer(buffer, value, nil); err != nil {
+		return err
+	}
+	if f.compatible && f.metaContext != nil && len(f.metaContext.writingTypeDefs) > 0 {
+		buffer.PutInt32(metaOffset, int32(buffer.writerIndex-metaOffset-4))
+		f.typeResolver.writeTypeDefs(buffer)
+	}
+	return nil
 }
 
 func (f *Fory) writeReferencableBySerializer(buffer *ByteBuffer, value reflect.Value, serializer Serializer) error {
@@ -367,6 +437,19 @@ func (f *Fory) Deserialize(buf *ByteBuffer, v interface{}, buffers []*ByteBuffer
 				"produced with buffer_callback null")
 		}
 	}
+
+	if f.compatible {
+		typeDefOffset := buf.ReadInt32()
+		if typeDefOffset >= 0 {
+			save := buf.readerIndex
+			buf.SetReaderIndex(save + int(typeDefOffset))
+			if err := f.typeResolver.readTypeDefs(buf); err != nil {
+				return fmt.Errorf("failed to read typeDefs: %w", err)
+			}
+			buf.SetReaderIndex(save)
+		}
+	}
+
 	if isXLangFlag {
 		return f.ReadReferencable(buf, reflect.ValueOf(v).Elem())
 	} else {
@@ -475,11 +558,25 @@ func (f *Fory) Reset() {
 func (f *Fory) resetWrite() {
 	f.typeResolver.resetWrite()
 	f.refResolver.resetWrite()
+	if f.metaContext != nil {
+		if f.metaContext.IsScopedMetaShareEnabled() {
+			f.metaContext.resetWrite()
+		} else {
+			f.metaContext = nil
+		}
+	}
 }
 
 func (f *Fory) resetRead() {
 	f.typeResolver.resetRead()
 	f.refResolver.resetRead()
+	if f.metaContext != nil {
+		if f.metaContext.IsScopedMetaShareEnabled() {
+			f.metaContext.resetRead()
+		} else {
+			f.metaContext = nil
+		}
+	}
 }
 
 // methods for configure fory.
@@ -488,6 +585,6 @@ func (f *Fory) SetLanguage(language Language) {
 	f.language = language
 }
 
-func (f *Fory) SetReferenceTracking(referenceTracking bool) {
-	f.referenceTracking = referenceTracking
+func (f *Fory) SetRefTracking(refTracking bool) {
+	f.refTracking = refTracking
 }