fix(java): fix scala object type codegen (#1659)

## What does this PR do?

fix scala object type codegen

## Related issues

Closes #1658

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

<!--
If any user-facing interface changes, please [open an
issue](https://github.com/apache/incubator-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/CodegenContext.java

@@ -252,8 +252,7 @@ public String namePrefix(Class<?> clz) {
     if (clz.isArray()) {
       return "arr";
     } else {
-      String canonicalName = clz.getCanonicalName();
-      String type = canonicalName != null ? type(clz) : "Object";
+      String type = clz.getCanonicalName() != null ? type(clz) : "Object";
       int index = type.lastIndexOf(".");
       String name;
       if (index >= 0) {
@@ -283,7 +282,7 @@ public String type(Class<?> clz) {
     if (clz.isArray()) {
       return getArrayType(clz);
     }
-    String type = ReflectionUtils.getCanonicalName(clz);
+    String type = ReflectionUtils.getLiteralName(clz);
     if (type.startsWith("java.lang")) {
       if (!type.substring("java.lang.".length()).contains(".")) {
         String simpleName = clz.getSimpleName();
@@ -360,7 +359,7 @@ public LinkedHashSet<String> getImports() {
    */
   public void addImports(Class<?>... classes) {
     for (Class<?> clz : classes) {
-      imports.add(ReflectionUtils.getCanonicalName(clz));
+      imports.add(ReflectionUtils.getLiteralName(clz));
     }
   }
 
@@ -382,7 +381,7 @@ public void addImports(String... imports) {
    * @param cls class to be imported
    */
   public void addImport(Class<?> cls) {
-    this.imports.add(ReflectionUtils.getCanonicalName(cls));
+    this.imports.add(ReflectionUtils.getLiteralName(cls));
   }
 
   /**

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

@@ -343,7 +343,7 @@ public ExprCode doGenCode(CodegenContext ctx) {
           if (valueClass.isArray()) {
             v = String.format("%s.class", TypeUtils.getArrayType((Class<?>) value));
           } else {
-            v = String.format("%s.class", ReflectionUtils.getCanonicalName((Class<?>) (value)));
+            v = String.format("%s.class", ReflectionUtils.getLiteralName((Class<?>) (value)));
           }
           return new ExprCode(FalseLiteral, new LiteralValue(javaType, v));
         } else {

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

@@ -554,17 +554,22 @@ public static String getPackage(String className) {
   }
 
   /**
-   * Returns the canonical name of the underlying class as defined by <cite>The Java Language
+   * Returns the literal name of the underlying class as defined by <cite>The Java Language
    * Specification</cite>. Throw {@link IllegalArgumentException} if the underlying class does not
    * have a canonical name(i.e., if it is a local or anonymous class or an array whose component
    * type does not have a canonical name).
    *
    * @throws IllegalArgumentException if the canonical name of the underlying class doesn't exist.
    */
-  public static String getCanonicalName(Class<?> cls) {
+  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);
+    if (canonicalName.endsWith(".")) {
+      // qualifier name of scala object type will ends with `.`
+      String name = cls.getName();
+      canonicalName = name.substring(0, name.length() - 1).replace("$", ".") + "$";
+    }
     return canonicalName;
   }
 

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

@@ -401,7 +401,7 @@ public static String getArrayType(TypeRef<?> type) {
   /** Returns s string that represents array type declaration of type. */
   public static String getArrayType(Class<?> type) {
     Tuple2<Class<?>, Integer> info = getArrayComponentInfo(type);
-    StringBuilder typeBuilder = new StringBuilder(ReflectionUtils.getCanonicalName(info.f0));
+    StringBuilder typeBuilder = new StringBuilder(ReflectionUtils.getLiteralName(info.f0));
     for (int i = 0; i < info.f1; i++) {
       typeBuilder.append("[]");
     }
@@ -410,7 +410,7 @@ 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.getCanonicalName(elemType));
+    StringBuilder typeBuilder = new StringBuilder(ReflectionUtils.getLiteralName(elemType));
     for (int i = 0; i < dimensions.length; i++) {
       typeBuilder.append('[').append(dimensions[i]).append(']');
     }

java/fury-testsuite/pom.xml

@@ -99,6 +99,12 @@
       <version>8.5.6</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-sql_2.13</artifactId>
+      <version>3.2.1</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

java/fury-testsuite/src/test/java/org/apache/fury/test/SparkTypeTest.java

@@ -0,0 +1,35 @@
+/*
+ * 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.test;
+
+import org.apache.fury.Fury;
+import org.apache.fury.TestBase;
+import org.apache.spark.sql.types.DecimalType;
+import org.apache.spark.sql.types.DecimalType$;
+import org.testng.annotations.Test;
+
+public class SparkTypeTest extends TestBase {
+  @Test(dataProvider = "enableCodegen")
+  public void testObjectType(boolean enableCodegen) {
+    Fury fury = builder().withRefTracking(true).withCodegen(enableCodegen).build();
+    fury.serialize(DecimalType$.MODULE$);
+    fury.serialize(new DecimalType(10, 10));
+  }
+}

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

@@ -40,4 +40,4 @@ class SingleObjectSerializerTest extends AnyWordSpec with Matchers {
       fury.deserialize(fury.serialize(Pair(singleton, singleton))) shouldEqual Pair(singleton, singleton)
     }
   }
-}
+}