feat(java): enable scoped meta share for compatible mode by default (#…

…1733)

## What does this PR do?

This PR enable scoped meta share for compatible mode by default:
- Enable scoped meta share for compatible mode by default
- Extensive tests for scoped meta share mode
- Support for graalvm

## Related issues

<!--
Is there any related issue? Please attach here.

- #xxxx0
- #xxxx1
- #xxxx2
-->


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

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

- [ ] 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.
-->

integration_tests/graalvm_tests/src/main/java/org/apache/fury/graalvm/CompatibleExample.java

@@ -30,6 +30,7 @@ public class CompatibleExample {
         Fury.builder()
             .requireClassRegistration(true)
             .withCompatibleMode(CompatibleMode.COMPATIBLE)
+            .withScopedMetaShare(false)
             .build();
     // register and generate serializer code.
     fury.register(Foo.class, true);

integration_tests/graalvm_tests/src/main/java/org/apache/fury/graalvm/Main.java

@@ -27,6 +27,7 @@ public class Main {
   public static void main(String[] args) throws Throwable {
     Example.main(args);
     CompatibleExample.main(args);
+    ScopedCompatibleExample.main(args);
     RecordExample.main(args);
     CompatibleRecordExample.main(args);
     RecordExample2.main(args);

integration_tests/graalvm_tests/src/main/java/org/apache/fury/graalvm/ScopedCompatibleExample.java

@@ -0,0 +1,43 @@
+/*
+ * 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 org.apache.fury.graalvm;
+
+import org.apache.fury.Fury;
+import org.apache.fury.config.CompatibleMode;
+
+public class ScopedCompatibleExample {
+  static Fury fury;
+
+  static {
+    fury =
+        Fury.builder()
+            .requireClassRegistration(true)
+            .withCompatibleMode(CompatibleMode.COMPATIBLE)
+            .withScopedMetaShare(true)
+            .build();
+    // register and generate serializer code.
+    fury.register(Foo.class, true);
+  }
+
+  public static void main(String[] args) {
+    Example.test(fury);
+    System.out.println("CompatibleExample succeed");
+  }
+}

integration_tests/latest_jdk_tests/src/test/java/org/apache/fury/integration_tests/RecordSerializersTest.java

@@ -111,11 +111,12 @@ public void testRecordCompatible(boolean codegen) throws Throwable {
         Fury.builder()
             .requireClassRegistration(false)
             .withCodegen(codegen)
+            .withClassLoader(cls1.getClassLoader())
             .withCompatibleMode(CompatibleMode.COMPATIBLE)
             .build();
     byte[] bytes1 = fury.serialize(record1);
     Object o = fury.deserialize(bytes1);
-    Assert.assertEquals(record1, o);
+    Assert.assertEquals(o, record1);
     String code2 =
         "import java.util.*;"
             + "public record TestRecord(int f1, String f2, char f4, Map<String, Integer> f5) {}";
@@ -128,10 +129,11 @@ public void testRecordCompatible(boolean codegen) throws Throwable {
         Fury.builder()
             .requireClassRegistration(false)
             .withCodegen(codegen)
+            .withClassLoader(cls2.getClassLoader())
             .withCompatibleMode(CompatibleMode.COMPATIBLE)
             .build();
     Object o2 = fury2.deserialize(fury2.serialize(record2));
-    Assert.assertEquals(record2, o2);
+    Assert.assertEquals(o2, record2);
     // test compatible
     Assert.assertEquals(fury2.deserialize(bytes1), record2);
   }

java/benchmark/src/test/java/org/apache/fury/benchmark/state/JsonTest.java

@@ -73,7 +73,6 @@ public void testSerializeJson(
             .withScopedMetaShare(scoped)
             .withCodegen(codegen)
             .registerGuavaTypes(false)
-            .withCompatibleMode(CompatibleMode.COMPATIBLE)
             .build();
     byte[] serialized = fury.serialize(resp);
     DemoResponse o = (DemoResponse) fury.deserialize(serialized);

java/fury-core/src/main/java/org/apache/fury/builder/MetaSharedCodecBuilder.java

@@ -41,6 +41,7 @@
 import org.apache.fury.type.Descriptor;
 import org.apache.fury.type.DescriptorGrouper;
 import org.apache.fury.util.ExceptionUtils;
+import org.apache.fury.util.GraalvmSupport;
 import org.apache.fury.util.Preconditions;
 import org.apache.fury.util.StringUtils;
 import org.apache.fury.util.record.RecordComponent;
@@ -136,6 +137,10 @@ protected void addCommonImports() {
   @SuppressWarnings({"unchecked", "rawtypes"})
   public static Serializer setCodegenSerializer(
       Fury fury, Class<?> cls, GeneratedMetaSharedSerializer s) {
+    if (GraalvmSupport.isGraalRuntime()) {
+      return fury.getJITContext()
+          .asyncVisitFury(f -> f.getClassResolver().getSerializer(s.getType()));
+    }
     // This method hold jit lock, so create jit serializer async to avoid block serialization.
     Class serializerClass =
         fury.getJITContext()

java/fury-core/src/main/java/org/apache/fury/config/FuryBuilder.java

@@ -70,8 +70,8 @@ public final class FuryBuilder {
   boolean checkJdkClassSerializable = true;
   Class<? extends Serializer> defaultJDKStreamSerializerType = ObjectStreamSerializer.class;
   boolean requireClassRegistration = true;
-  boolean metaShareEnabled = false;
-  boolean scopedMetaShareEnabled = false;
+  Boolean metaShareEnabled;
+  Boolean scopedMetaShareEnabled;
   boolean codeGenEnabled = true;
   Boolean deserializeNonexistentClass;
   boolean asyncCompilationEnabled = false;
@@ -239,6 +239,9 @@ public FuryBuilder suppressClassRegistrationWarnings(boolean suppress) {
   /** Whether to enable meta share mode. */
   public FuryBuilder withMetaShare(boolean shareMeta) {
     this.metaShareEnabled = shareMeta;
+    if (!shareMeta) {
+      scopedMetaShareEnabled = false;
+    }
     return this;
   }
 
@@ -248,9 +251,6 @@ public FuryBuilder withMetaShare(boolean shareMeta) {
    */
   public FuryBuilder withScopedMetaShare(boolean scoped) {
     scopedMetaShareEnabled = scoped;
-    if (scoped) {
-      metaShareEnabled = true;
-    }
     return this;
   }
 
@@ -330,10 +330,29 @@ private void finish() {
       if (deserializeNonexistentClass == null) {
         deserializeNonexistentClass = true;
       }
+      if (scopedMetaShareEnabled == null) {
+        if (metaShareEnabled == null) {
+          metaShareEnabled = true;
+          scopedMetaShareEnabled = true;
+        } else {
+          scopedMetaShareEnabled = false;
+        }
+      } else {
+        if (metaShareEnabled == null) {
+          metaShareEnabled = scopedMetaShareEnabled;
+        }
+      }
     } else {
       if (deserializeNonexistentClass == null) {
         deserializeNonexistentClass = false;
       }
+      if (scopedMetaShareEnabled != null) {
+        LOG.warn("Scoped meta share is for CompatibleMode only, disable it for {}", compatibleMode);
+      }
+      scopedMetaShareEnabled = false;
+      if (metaShareEnabled == null) {
+        metaShareEnabled = false;
+      }
     }
     if (!requireClassRegistration) {
       LOG.warn(

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

@@ -79,7 +79,9 @@ public static ClassDef decodeClassDef(ClassResolver classResolver, MemoryBuffer
       int numFields = currentClassHeader >>> 1;
       if (isRegistered) {
         int registeredId = classDefBuf.readVarUint32Small7();
-        className = classResolver.getClassInfo((short) registeredId).getCls().getName();
+        Class<?> cls = classResolver.getClassInfo((short) registeredId).getCls();
+        className = cls.getName();
+        classSpec = new ClassSpec(cls);
       } else {
         String pkg = readPkgName(classDefBuf);
         String typeName = readTypeName(classDefBuf);

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

@@ -19,6 +19,8 @@
 
 package org.apache.fury.meta;
 
+import org.apache.fury.type.TypeUtils;
+
 public class ClassSpec {
   public final String entireClassName;
 
@@ -28,6 +30,13 @@ public class ClassSpec {
   public final boolean isArray;
   public final int dimension;
 
+  public ClassSpec(Class<?> cls) {
+    this.entireClassName = cls.getName();
+    isEnum = cls.isEnum();
+    isArray = cls.isArray();
+    dimension = isArray ? TypeUtils.getArrayDimensions(cls) : 0;
+  }
+
   public ClassSpec(String entireClassName, boolean isEnum, boolean isArray, int dimension) {
     this.entireClassName = entireClassName;
     this.isEnum = isEnum;

java/fury-core/src/main/java/org/apache/fury/reflect/ReflectionUtils.java

@@ -34,6 +34,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
@@ -696,8 +697,8 @@ private static boolean objectCommonFieldsEquals(
       Object o2) {
 
     for (String commonField : commonFieldsInfo.f0) {
-      Field field1 = commonFieldsInfo.f1.get(commonField);
-      Field field2 = commonFieldsInfo.f2.get(commonField);
+      Field field1 = Objects.requireNonNull(commonFieldsInfo.f1.get(commonField));
+      Field field2 = Objects.requireNonNull(commonFieldsInfo.f2.get(commonField));
       FieldAccessor accessor1 = FieldAccessor.createAccessor(field1);
       FieldAccessor accessor2 = FieldAccessor.createAccessor(field2);
       Object f1 = accessor1.get(o1);
@@ -769,15 +770,13 @@ public static Tuple3<Set<String>, Map<String, Field>, Map<String, Field>> getCom
             .collect(
                 Collectors.toMap(
                     // don't use `getGenericType` since janino doesn't support generics.
-                    f -> f.getDeclaringClass().getSimpleName() + f.getType() + f.getName(),
+                    f -> f.getDeclaringClass().getSimpleName() + f.getName(),
                     f -> f));
     Map<String, Field> fieldMap2 =
         fields2.stream()
             .collect(
-                Collectors.toMap(
-                    f -> f.getDeclaringClass().getSimpleName() + f.getType() + f.getName(),
-                    f -> f));
-    Set<String> commonFields = fieldMap1.keySet();
+                Collectors.toMap(f -> f.getDeclaringClass().getSimpleName() + f.getName(), f -> f));
+    Set<String> commonFields = new HashSet<>(fieldMap1.keySet());
     commonFields.retainAll(fieldMap2.keySet());
     return Tuple3.of(commonFields, fieldMap1, fieldMap2);
   }

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

@@ -24,6 +24,7 @@
 
 import org.apache.fury.collection.Tuple2;
 import org.apache.fury.config.Language;
+import org.apache.fury.meta.ClassDef;
 import org.apache.fury.meta.Encoders;
 import org.apache.fury.meta.MetaString.Encoding;
 import org.apache.fury.reflect.ReflectionUtils;
@@ -36,7 +37,6 @@
  * serialization.
  */
 public class ClassInfo {
-
   final Class<?> cls;
   final MetaStringBytes fullClassNameBytes;
   final MetaStringBytes packageNameBytes;
@@ -47,6 +47,8 @@ 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;
+  ClassDef classDef;
+  public boolean needToWriteClassDef;
 
   ClassInfo(
       Class<?> cls,
@@ -78,6 +80,7 @@ public class ClassInfo {
       short classId) {
     this.cls = cls;
     this.serializer = serializer;
+    needToWriteClassDef = serializer != null && classResolver.needToWriteClassDef(serializer);
     MetaStringResolver metaStringResolver = classResolver.getMetaStringResolver();
     if (cls != null && classResolver.getFury().getLanguage() != Language.JAVA) {
       this.fullClassNameBytes =
@@ -90,9 +93,7 @@ public class ClassInfo {
     // means only classes are serialized, not the instance. If we
     // serialize such class only, we need to write classname bytes.
     if (cls != null
-        && ((classId == ClassResolver.NO_CLASS_ID
-                && !classResolver.getFury().getConfig().isMetaShareEnabled())
-            || classId == ClassResolver.REPLACE_STUB_ID)) {
+        && (classId == ClassResolver.NO_CLASS_ID || classId == ClassResolver.REPLACE_STUB_ID)) {
       // REPLACE_STUB_ID for write replace class in `ClassSerializer`.
       Tuple2<String, String> tuple2 = Encoders.encodePkgAndClass(cls);
       this.packageNameBytes =
@@ -147,6 +148,11 @@ public <T> Serializer<T> getSerializer() {
     return (Serializer<T>) serializer;
   }
 
+  void setSerializer(ClassResolver resolver, Serializer<?> serializer) {
+    this.serializer = serializer;
+    needToWriteClassDef = serializer != null && resolver.needToWriteClassDef(serializer);
+  }
+
   @Override
   public String toString() {
     return "ClassInfo{"