fix(scala): fix nested type serialization in scala object type (#1809)

## What does this PR do?

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


## Related issues

Closes #1801 


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

java/fury-core/src/main/java/org/apache/fury/codegen/CodeGenerator.java

@@ -496,9 +496,15 @@ public static boolean sourcePkgLevelAccessible(Class<?> clz) {
     if (clz.isPrimitive()) {
       return true;
     }
-    if (clz.getCanonicalName() == null) {
+    try {
+      if (clz.getCanonicalName() == null) {
+        return false;
+      }
+    } catch (InternalError e) {
+      // getCanonicalName for scala type `A$B$C` may fail
       return false;
     }
+
     // Scala may produce class name like: xxx.SomePackageObject.package$SomeClass
     HashSet<String> set = Collections.ofHashSet(clz.getCanonicalName().split("\\."));
     return !Collections.hasIntersection(set, CodegenContext.JAVA_RESERVED_WORDS);

java/fury-core/src/main/java/org/apache/fury/codegen/CodegenContext.java

@@ -42,6 +42,8 @@
 import org.apache.fury.codegen.Expression.Reference;
 import org.apache.fury.collection.Collections;
 import org.apache.fury.collection.Tuple2;
+import org.apache.fury.logging.Logger;
+import org.apache.fury.logging.LoggerFactory;
 import org.apache.fury.reflect.ReflectionUtils;
 import org.apache.fury.reflect.TypeRef;
 import org.apache.fury.util.Preconditions;
@@ -55,6 +57,8 @@
  * constructor's args.
  */
 public class CodegenContext {
+  private static final Logger LOG = LoggerFactory.getLogger(CodegenContext.class);
+
   public static Set<String> JAVA_RESERVED_WORDS;
 
   static {
@@ -252,7 +256,13 @@ public String namePrefix(Class<?> clz) {
     if (clz.isArray()) {
       return "arr";
     } else {
-      String type = clz.getCanonicalName() != null ? type(clz) : "Object";
+      String type;
+      try {
+        // getCanonicalName for scala type `A$B$C` may fail
+        type = clz.getCanonicalName() != null ? type(clz) : "object";
+      } catch (InternalError e) {
+        type = "object";
+      }
       int index = type.lastIndexOf(".");
       String name;
       if (index >= 0) {

java/fury-core/src/main/java/org/apache/fury/codegen/Expression.java

@@ -341,9 +341,9 @@ public ExprCode doGenCode(CodegenContext ctx) {
           String v;
           Class<?> valueClass = (Class<?>) value;
           if (valueClass.isArray()) {
-            v = String.format("%s.class", TypeUtils.getArrayType((Class<?>) value));
+            v = String.format("%s.class", TypeUtils.getArrayClass((Class<?>) value));
           } else {
-            v = String.format("%s.class", ReflectionUtils.getLiteralName((Class<?>) (value)));
+            v = String.format("%s.class", ((Class<?>) (value)).getName());
           }
           return new ExprCode(FalseLiteral, new LiteralValue(javaType, v));
         } else {
@@ -1273,10 +1273,10 @@ public ExprCode doGenCode(CodegenContext ctx) {
       }
 
       Class<?> rawType = getRawType(type);
-      String type = ctx.type(rawType);
+      String typename = ctx.type(rawType);
       String clzName = unknownClassName;
       if (clzName == null) {
-        clzName = type;
+        clzName = rawType.getName();
       }
       if (needOuterPointer) {
         // "${gen.value}.new ${cls.getSimpleName}($argString)"
@@ -1298,9 +1298,9 @@ public ExprCode doGenCode(CodegenContext ctx) {
           codeBuilder.append(code).append('\n');
           String cast =
               StringUtils.format(
-                  "${clzName} ${value} = (${clzName})${instance};",
-                  "clzName",
-                  clzName,
+                  "${type} ${value} = (${type})${instance};",
+                  "type",
+                  typename,
                   "value",
                   value,
                   "instance",
@@ -1309,9 +1309,9 @@ public ExprCode doGenCode(CodegenContext ctx) {
         } else {
           String code =
               StringUtils.format(
-                  "${clzName} ${value} = new ${clzName}(${args});",
-                  "clzName",
-                  clzName,
+                  "${type} ${value} = new ${type}(${args});",
+                  "type",
+                  ReflectionUtils.isAbstract(rawType) ? clzName : typename,
                   "value",
                   value,
                   "args",

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

@@ -41,6 +41,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.WeakHashMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.stream.Collectors;
@@ -566,13 +567,38 @@ public static String getPackage(String className) {
    * @throws IllegalArgumentException if the canonical name of the underlying class doesn't exist.
    */
   public static String getLiteralName(Class<?> cls) {
-    String canonicalName = cls.getCanonicalName();
-    org.apache.fury.util.Preconditions.checkArgument(
-        canonicalName != null, "Class %s doesn't have canonical name", cls);
+    String canonicalName;
+    try {
+      // getCanonicalName for scala type `A$B$C` may fail
+      canonicalName = cls.getCanonicalName();
+    } catch (InternalError e) {
+      return cls.getName();
+    }
+    if (canonicalName == null) {
+      throw new NullPointerException(String.format("Class %s doesn't have canonical name", cls));
+    }
+    String clsName = cls.getName();
     if (canonicalName.endsWith(".")) {
       // qualifier name of scala object type will ends with `.`
-      String name = cls.getName();
-      canonicalName = name.substring(0, name.length() - 1).replace("$", ".") + "$";
+      canonicalName = clsName.substring(0, clsName.length() - 1).replace("$", ".") + "$";
+    } else {
+      if (!canonicalName.endsWith("$") && canonicalName.contains("$")) {
+        // nested scala object type can't be accessed in java by using canonicalName
+        // see more detailed in
+        // https://stackoverflow.com/questions/30809070/accessing-scala-nested-classes-from-java
+        int nestedLevels = 0;
+        boolean hasEnclosedObjectType = false;
+        while (cls.getEnclosingClass() != null) {
+          nestedLevels++;
+          cls = cls.getEnclosingClass();
+          if (!hasEnclosedObjectType) {
+            hasEnclosedObjectType = isScalaSingletonObject(cls);
+          }
+        }
+        if (nestedLevels >= 2 && hasEnclosedObjectType) {
+          canonicalName = clsName;
+        }
+      }
     }
     return canonicalName;
   }
@@ -790,13 +816,20 @@ public static boolean isDynamicGeneratedCLass(Class<?> cls) {
     return Functions.isLambda(cls) || isJdkProxy(cls);
   }
 
+  private static final WeakHashMap<Class<?>, Boolean> scalaSingletonObjectCache =
+      new WeakHashMap<>();
+
   /** Returns true if a class is a scala `object` singleton. */
   public static boolean isScalaSingletonObject(Class<?> cls) {
-    try {
-      cls.getDeclaredField("MODULE$");
-      return true;
-    } catch (NoSuchFieldException e) {
-      return false;
-    }
+    return scalaSingletonObjectCache.computeIfAbsent(
+        cls,
+        c -> {
+          try {
+            cls.getDeclaredField("MODULE$");
+            return true;
+          } catch (NoSuchFieldException e) {
+            return false;
+          }
+        });
   }
 }

java/fury-core/src/main/java/org/apache/fury/type/TypeUtils.java

@@ -411,8 +411,17 @@ public static String getArrayType(Class<?> type) {
   /** Create an array type declaration from elemType and dimensions. */
   public static String getArrayType(Class<?> elemType, int[] dimensions) {
     StringBuilder typeBuilder = new StringBuilder(ReflectionUtils.getLiteralName(elemType));
-    for (int i = 0; i < dimensions.length; i++) {
-      typeBuilder.append('[').append(dimensions[i]).append(']');
+    for (int dimension : dimensions) {
+      typeBuilder.append('[').append(dimension).append(']');
+    }
+    return typeBuilder.toString();
+  }
+
+  public static String getArrayClass(Class<?> type) {
+    Tuple2<Class<?>, Integer> info = getArrayComponentInfo(type);
+    StringBuilder typeBuilder = new StringBuilder(info.f0.getName());
+    for (int i = 0; i < info.f1; i++) {
+      typeBuilder.append("[]");
     }
     return typeBuilder.toString();
   }

scala/src/test/scala/org/apache/fury/serializer/SingleObjectSerializerTest.scala

@@ -28,6 +28,19 @@ object singleton {}
 
 case class Pair(f1: Any, f2: Any)
 
+object A {
+  object B {
+    case class C(value: String) {
+    }
+  }
+}
+
+class X {
+  class Y {
+    class Z
+  }
+}
+
 class SingleObjectSerializerTest extends AnyWordSpec with Matchers {
   "fury scala object support" should {
     "serialize/deserialize" in {
@@ -39,5 +52,15 @@ class SingleObjectSerializerTest extends AnyWordSpec with Matchers {
       fury.deserialize(fury.serialize(singleton)) shouldBe singleton
       fury.deserialize(fury.serialize(Pair(singleton, singleton))) shouldEqual Pair(singleton, singleton)
     }
+    "nested type serialization in object type" in {
+      val fury = Fury.builder()
+        .withLanguage(Language.JAVA)
+        .withRefTracking(true)
+        .withScalaOptimizationEnabled(true)
+        .requireClassRegistration(false).build()
+      val x = A.B.C("hello, world!")
+      val bytes = fury.serialize(x)
+      fury.deserialize(bytes) shouldEqual A.B.C("hello, world!")
+    }
   }
 }