feat(java): Type annotation hints for serialization(WIP)

java/benchmark/pom.xml

@@ -139,16 +139,6 @@
       <artifactId>smoothie-map</artifactId>
       <version>2.0.2</version>
     </dependency>
-    <dependency>
-      <groupId>com.google.protobuf</groupId>
-      <artifactId>protobuf-java</artifactId>
-      <version>${protobuf.version}</version>
-    </dependency>
-    <dependency>
-      <groupId>com.google.flatbuffers</groupId>
-      <artifactId>flatbuffers-java</artifactId>
-      <version>2.0.3</version>
-    </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>

java/fury-core/src/main/java/org/apache/fury/Fury.java

@@ -462,6 +462,29 @@ public <T> void writeRef(MemoryBuffer buffer, T obj, Serializer<T> serializer) {
     }
   }
 
+  public <T> void writeRefNullable(
+      MemoryBuffer buffer, T obj, Serializer<T> serializer, boolean nullable) {
+    if (serializer.needToWriteRef()) {
+      if (!refResolver.writeRefOrNull(buffer, obj)) {
+        depth++;
+        serializer.write(buffer, obj);
+        depth--;
+      }
+    } else {
+      if (nullable) {
+        if (obj == null) {
+          buffer.writeByte(Fury.NULL_FLAG);
+          return;
+        } else {
+          buffer.writeByte(Fury.NOT_NULL_VALUE_FLAG);
+        }
+      }
+      depth++;
+      serializer.write(buffer, obj);
+      depth--;
+    }
+  }
+
   /** Write object class and data without tracking ref. */
   public void writeNullable(MemoryBuffer buffer, Object obj) {
     if (obj == null) {
@@ -697,6 +720,25 @@ public void writeJavaStringRef(MemoryBuffer buffer, String str) {
     }
   }
 
+  public void writeJavaStringRef(MemoryBuffer buffer, String str, boolean nullable) {
+    if (stringSerializer.needToWriteRef()) {
+      if (!refResolver.writeRefOrNull(buffer, str)) {
+        stringSerializer.writeJavaString(buffer, str);
+      }
+    } else {
+      if (nullable) {
+        if (str == null) {
+          buffer.writeByte(Fury.NULL_FLAG);
+        } else {
+          buffer.writeByte(Fury.NOT_NULL_VALUE_FLAG);
+          stringSerializer.write(buffer, str);
+        }
+      } else {
+        stringSerializer.write(buffer, str);
+      }
+    }
+  }
+
   public String readJavaStringRef(MemoryBuffer buffer) {
     RefResolver refResolver = this.refResolver;
     if (stringSerializer.needToWriteRef()) {
@@ -719,6 +761,32 @@ public String readJavaStringRef(MemoryBuffer buffer) {
     }
   }
 
+  public String readJavaStringRef(MemoryBuffer buffer, boolean nullable) {
+    RefResolver refResolver = this.refResolver;
+    if (stringSerializer.needToWriteRef()) {
+      String obj;
+      int nextReadRefId = refResolver.tryPreserveRefId(buffer);
+      if (nextReadRefId >= NOT_NULL_VALUE_FLAG) {
+        obj = stringSerializer.read(buffer);
+        refResolver.setReadObject(nextReadRefId, obj);
+        return obj;
+      } else {
+        return (String) refResolver.getReadObject();
+      }
+    } else {
+      if (nullable) {
+        byte headFlag = buffer.readByte();
+        if (headFlag == Fury.NULL_FLAG) {
+          return null;
+        } else {
+          return stringSerializer.read(buffer);
+        }
+      } else {
+        return stringSerializer.read(buffer);
+      }
+    }
+  }
+
   public void writeJavaString(MemoryBuffer buffer, String str) {
     stringSerializer.writeJavaString(buffer, str);
   }
@@ -916,6 +984,29 @@ public <T> T readRef(MemoryBuffer buffer, Serializer<T> serializer) {
     }
   }
 
+  @SuppressWarnings("unchecked")
+  public <T> T readRefNullable(MemoryBuffer buffer, Serializer<T> serializer, boolean nullable) {
+    if (serializer.needToWriteRef()) {
+      T obj;
+      int nextReadRefId = refResolver.tryPreserveRefId(buffer);
+      if (nextReadRefId >= NOT_NULL_VALUE_FLAG) {
+        obj = serializer.read(buffer);
+        refResolver.setReadObject(nextReadRefId, obj);
+        return obj;
+      } else {
+        return (T) refResolver.getReadObject();
+      }
+    } else {
+      if (nullable) {
+        byte headFlag = buffer.readByte();
+        if (headFlag == Fury.NULL_FLAG) {
+          return null;
+        }
+      }
+      return serializer.read(buffer);
+    }
+  }
+
   /** Deserialize not-null and non-reference object from <code>buffer</code>. */
   public Object readNonRef(MemoryBuffer buffer) {
     return readDataInternal(buffer, classResolver.readClassInfo(buffer));
@@ -935,15 +1026,6 @@ public Object readNullable(MemoryBuffer buffer) {
     }
   }
 
-  public Object readNullable(MemoryBuffer buffer, ClassInfoHolder classInfoHolder) {
-    byte headFlag = buffer.readByte();
-    if (headFlag == Fury.NULL_FLAG) {
-      return null;
-    } else {
-      return readNonRef(buffer, classInfoHolder);
-    }
-  }
-
   /** Class should be read already. */
   public Object readData(MemoryBuffer buffer, ClassInfo classInfo) {
     depth++;

java/fury-core/src/main/java/org/apache/fury/annotation/FuryField.java

@@ -0,0 +1,33 @@
+/*
+ * 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.annotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.FIELD)
+public @interface FuryField {
+
+  /** Whether field is nullable. */
+  boolean nullable() default true;
+}

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

@@ -43,12 +43,14 @@
 import java.util.TreeSet;
 import java.util.stream.Collectors;
 import org.apache.fury.Fury;
+import org.apache.fury.annotation.FuryField;
 import org.apache.fury.collection.Tuple2;
 import org.apache.fury.exception.ClassNotCompatibleException;
 import org.apache.fury.memory.MemoryBuffer;
 import org.apache.fury.reflect.FieldAccessor;
 import org.apache.fury.reflect.ReflectionUtils;
 import org.apache.fury.reflect.TypeRef;
+import org.apache.fury.serializer.AbstractObjectSerializer;
 import org.apache.fury.serializer.PrimitiveSerializers;
 import org.apache.fury.serializer.collection.AbstractCollectionSerializer;
 import org.apache.fury.serializer.collection.AbstractMapSerializer;
@@ -748,6 +750,7 @@ public static class FieldInfo {
     protected final ClassResolver classResolver;
     private final FieldAccessor fieldAccessor;
     private final ClassInfoHolder classInfoHolder;
+    private final AbstractObjectSerializer.FuryFieldAnnotationInfo furyFieldAnnotationInfo;
 
     public FieldInfo(
         Fury fury,
@@ -772,6 +775,9 @@ public FieldInfo(
       } else {
         fieldAccessor = FieldAccessor.createAccessor(field);
       }
+      this.furyFieldAnnotationInfo =
+          new AbstractObjectSerializer.FuryFieldAnnotationInfo(
+              field == null ? null : field.getAnnotation(FuryField.class));
     }
 
     public static FieldInfo of(
@@ -838,6 +844,10 @@ public static FieldInfo of(
       }
     }
 
+    public AbstractObjectSerializer.FuryFieldAnnotationInfo getFuryFieldAnnotationInfo() {
+      return furyFieldAnnotationInfo;
+    }
+
     public String getName() {
       return name;
     }

java/fury-core/src/main/java/org/apache/fury/serializer/AbstractObjectSerializer.java

@@ -31,6 +31,7 @@
 import java.util.List;
 import java.util.stream.Collectors;
 import org.apache.fury.Fury;
+import org.apache.fury.annotation.FuryField;
 import org.apache.fury.collection.Tuple2;
 import org.apache.fury.collection.Tuple3;
 import org.apache.fury.memory.Platform;
@@ -376,7 +377,10 @@ protected T newBean() {
               descriptor.getField() != null
                   ? FieldAccessor.createAccessor(descriptor.getField())
                   : null,
-              fury);
+              fury,
+              descriptor.getField() != null
+                  ? descriptor.getField().getAnnotation(FuryField.class)
+                  : null);
       otherFields[cnt++] = genericTypeField;
     }
     cnt = 0;
@@ -398,27 +402,55 @@ private static FinalTypeField buildFinalTypeField(Fury fury, Descriptor d) {
         d.getDeclaringClass() + "." + d.getName(),
         // `d.getField()` will be null when peer class doesn't have this field.
         d.getField() != null ? FieldAccessor.createAccessor(d.getField()) : null,
-        fury);
+        fury,
+        d.getField() != null ? d.getField().getAnnotation(FuryField.class) : null);
   }
 
   private static GenericTypeField buildContainerField(Fury fury, Descriptor d) {
     return new GenericTypeField(
         d.getTypeRef(),
         d.getDeclaringClass() + "." + d.getName(),
         d.getField() != null ? FieldAccessor.createAccessor(d.getField()) : null,
-        fury);
+        fury,
+        d.getField() != null ? d.getField().getAnnotation(FuryField.class) : null);
+  }
+
+  /**
+   * This class is used to store the properties of the annotation {@link FuryField}, avoiding the
+   * reflection overhead when using annotations to obtain properties.
+   */
+  public static final class FuryFieldAnnotationInfo {
+
+    // default value is true
+    public boolean nullable = true;
+
+    public FuryFieldAnnotationInfo(FuryField furyField) {
+      if (furyField != null) {
+        this.nullable = furyField.nullable();
+      }
+    }
+
+    @Override
+    public String toString() {
+      return "FuryFieldAnnotationInfo{" + "nullable=" + nullable + '}';
+    }
   }
 
   public static class InternalFieldInfo {
     protected final short classId;
     protected final String qualifiedFieldName;
     protected final FieldAccessor fieldAccessor;
+    protected FuryFieldAnnotationInfo fieldAnnotationInfo;
 
     private InternalFieldInfo(
-        short classId, String qualifiedFieldName, FieldAccessor fieldAccessor) {
+        short classId,
+        String qualifiedFieldName,
+        FieldAccessor fieldAccessor,
+        FuryField furyField) {
       this.classId = classId;
       this.qualifiedFieldName = qualifiedFieldName;
       this.fieldAccessor = fieldAccessor;
+      this.fieldAnnotationInfo = new FuryFieldAnnotationInfo(furyField);
     }
 
     @Override
@@ -430,15 +462,18 @@ public String toString() {
           + qualifiedFieldName
           + ", field="
           + (fieldAccessor != null ? fieldAccessor.getField() : null)
+          + ", FuryFieldAnnotationInfo="
+          + fieldAnnotationInfo
           + '}';
     }
   }
 
   static final class FinalTypeField extends InternalFieldInfo {
     final ClassInfo classInfo;
 
-    private FinalTypeField(Class<?> type, String fieldName, FieldAccessor accessor, Fury fury) {
-      super(getRegisteredClassId(fury, type), fieldName, accessor);
+    private FinalTypeField(
+        Class<?> type, String fieldName, FieldAccessor accessor, Fury fury, FuryField furyField) {
+      super(getRegisteredClassId(fury, type), fieldName, accessor, furyField);
       // invoke `copy` to avoid ObjectSerializer construct clear serializer by `clearSerializer`.
       if (type == FinalObjectTypeStub.class) {
         // `FinalObjectTypeStub` has no fields, using its `classInfo`
@@ -456,17 +491,26 @@ static final class GenericTypeField extends InternalFieldInfo {
     final boolean trackingRef;
 
     private GenericTypeField(
-        Class<?> cls, String qualifiedFieldName, FieldAccessor accessor, Fury fury) {
-      super(getRegisteredClassId(fury, cls), qualifiedFieldName, accessor);
+        Class<?> cls,
+        String qualifiedFieldName,
+        FieldAccessor accessor,
+        Fury fury,
+        FuryField furyField) {
+      super(getRegisteredClassId(fury, cls), qualifiedFieldName, accessor, furyField);
       // TODO support generics <T> in Pojo<T>, see ComplexObjectSerializer.getGenericTypes
       genericType = fury.getClassResolver().buildGenericType(cls);
       classInfoHolder = fury.getClassResolver().nilClassInfoHolder();
       trackingRef = fury.getClassResolver().needToWriteRef(cls);
     }
 
     private GenericTypeField(
-        TypeRef<?> typeRef, String qualifiedFieldName, FieldAccessor accessor, Fury fury) {
-      super(getRegisteredClassId(fury, getRawType(typeRef)), qualifiedFieldName, accessor);
+        TypeRef<?> typeRef,
+        String qualifiedFieldName,
+        FieldAccessor accessor,
+        Fury fury,
+        FuryField furyField) {
+      super(
+          getRegisteredClassId(fury, getRawType(typeRef)), qualifiedFieldName, accessor, furyField);
       // TODO support generics <T> in Pojo<T>, see ComplexObjectSerializer.getGenericTypes
       genericType = fury.getClassResolver().buildGenericType(typeRef);
       classInfoHolder = fury.getClassResolver().nilClassInfoHolder();
@@ -484,6 +528,8 @@ public String toString() {
           + qualifiedFieldName
           + ", field="
           + (fieldAccessor != null ? fieldAccessor.getField() : null)
+          + ", FuryFieldAnnotationInfo="
+          + fieldAnnotationInfo
           + '}';
     }
   }