feat(java): add deserialize overloads that take a TypeRef (#3946)

java/fory-core/src/main/java/org/apache/fory/BaseFory.java

@@ -25,6 +25,7 @@
 import org.apache.fory.io.ForyInputStream;
 import org.apache.fory.io.ForyReadableChannel;
 import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.reflect.TypeRef;
 import org.apache.fory.resolver.TypeResolver;
 import org.apache.fory.serializer.BufferCallback;
 import org.apache.fory.serializer.Serializer;
@@ -234,6 +235,14 @@ void registerSerializerAndType(
 
   <T> T deserialize(ForyReadableChannel channel, Class<T> type);
 
+  <T> T deserialize(byte[] bytes, TypeRef<T> typeRef);
+
+  <T> T deserialize(MemoryBuffer buffer, TypeRef<T> typeRef);
+
+  <T> T deserialize(ForyInputStream inputStream, TypeRef<T> typeRef);
+
+  <T> T deserialize(ForyReadableChannel channel, TypeRef<T> typeRef);
+
   Object deserialize(byte[] bytes, Iterable<MemoryBuffer> outOfBandBuffers);
 
   /** Deserialize <code>obj</code> from a <code>buffer</code>. */

java/fory-core/src/main/java/org/apache/fory/Fory.java

@@ -51,6 +51,7 @@
 import org.apache.fory.logging.LoggerFactory;
 import org.apache.fory.memory.MemoryBuffer;
 import org.apache.fory.memory.MemoryUtils;
+import org.apache.fory.reflect.TypeRef;
 import org.apache.fory.resolver.ClassResolver;
 import org.apache.fory.resolver.SharedRegistry;
 import org.apache.fory.resolver.TypeChecker;
@@ -462,6 +463,34 @@ public <T> T deserialize(ForyReadableChannel channel, Class<T> type) {
     return deserialize(channel.getBuffer(), type);
   }
 
+  @SuppressWarnings("unchecked")
+  @Override
+  public <T> T deserialize(byte[] bytes, TypeRef<T> typeRef) {
+    Preconditions.checkNotNull(typeRef, "typeRef must not be null");
+    return deserialize(bytes, (Class<T>) typeRef.getRawType());
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  public <T> T deserialize(MemoryBuffer buffer, TypeRef<T> typeRef) {
+    Preconditions.checkNotNull(typeRef, "typeRef must not be null");
+    return deserialize(buffer, (Class<T>) typeRef.getRawType());
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  public <T> T deserialize(ForyInputStream inputStream, TypeRef<T> typeRef) {
+    Preconditions.checkNotNull(typeRef, "typeRef must not be null");
+    return deserialize(inputStream, (Class<T>) typeRef.getRawType());
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  public <T> T deserialize(ForyReadableChannel channel, TypeRef<T> typeRef) {
+    Preconditions.checkNotNull(typeRef, "typeRef must not be null");
+    return deserialize(channel, (Class<T>) typeRef.getRawType());
+  }
+
   @Override
   public Object deserialize(byte[] bytes, Iterable<MemoryBuffer> outOfBandBuffers) {
     return deserialize(MemoryUtils.wrap(bytes), outOfBandBuffers);

java/fory-core/src/main/java/org/apache/fory/ThreadLocalFory.java

@@ -33,6 +33,7 @@
 import org.apache.fory.io.ForyInputStream;
 import org.apache.fory.io.ForyReadableChannel;
 import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.reflect.TypeRef;
 import org.apache.fory.resolver.SharedRegistry;
 import org.apache.fory.serializer.BufferCallback;
 
@@ -146,6 +147,26 @@ public <T> T deserialize(ForyReadableChannel channel, Class<T> type) {
     return currentFory().deserialize(channel, type);
   }
 
+  @Override
+  public <T> T deserialize(byte[] bytes, TypeRef<T> typeRef) {
+    return currentFory().deserialize(bytes, typeRef);
+  }
+
+  @Override
+  public <T> T deserialize(MemoryBuffer buffer, TypeRef<T> typeRef) {
+    return currentFory().deserialize(buffer, typeRef);
+  }
+
+  @Override
+  public <T> T deserialize(ForyInputStream inputStream, TypeRef<T> typeRef) {
+    return currentFory().deserialize(inputStream, typeRef);
+  }
+
+  @Override
+  public <T> T deserialize(ForyReadableChannel channel, TypeRef<T> typeRef) {
+    return currentFory().deserialize(channel, typeRef);
+  }
+
   @Override
   public Object deserialize(byte[] bytes, Iterable<MemoryBuffer> outOfBandBuffers) {
     return currentFory().deserialize(bytes, outOfBandBuffers);

java/fory-core/src/main/java/org/apache/fory/io/BlockedStreamUtils.java

@@ -29,6 +29,7 @@
 import org.apache.fory.Fory;
 import org.apache.fory.exception.DeserializationException;
 import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.reflect.TypeRef;
 import org.apache.fory.serializer.BufferCallback;
 import org.apache.fory.util.ExceptionUtils;
 
@@ -82,6 +83,16 @@ public static <T> T deserialize(Fory fory, ReadableByteChannel channel, Class<T>
     return type.cast(readFromChannel(fory, channel, b -> fory.deserialize(b, type)));
   }
 
+  @SuppressWarnings("unchecked")
+  public static <T> T deserialize(Fory fory, InputStream inputStream, TypeRef<T> typeRef) {
+    return (T) deserializeFromStream(fory, inputStream, buf -> fory.deserialize(buf, typeRef));
+  }
+
+  @SuppressWarnings("unchecked")
+  public static <T> T deserialize(Fory fory, ReadableByteChannel channel, TypeRef<T> typeRef) {
+    return (T) readFromChannel(fory, channel, b -> fory.deserialize(b, typeRef));
+  }
+
   private static Object readFromChannel(
       Fory fory, ReadableByteChannel channel, Function<MemoryBuffer, Object> action) {
     try {

java/fory-core/src/main/java/org/apache/fory/pool/ThreadPoolFory.java

@@ -35,6 +35,7 @@
 import org.apache.fory.io.ForyInputStream;
 import org.apache.fory.io.ForyReadableChannel;
 import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.reflect.TypeRef;
 import org.apache.fory.resolver.SharedRegistry;
 import org.apache.fory.serializer.BufferCallback;
 
@@ -274,6 +275,46 @@ public <T> T deserialize(ForyReadableChannel channel, Class<T> type) {
     }
   }
 
+  @Override
+  public <T> T deserialize(byte[] bytes, TypeRef<T> typeRef) {
+    PooledEntry entry = acquire();
+    try {
+      return entry.fory.deserialize(bytes, typeRef);
+    } finally {
+      release(entry);
+    }
+  }
+
+  @Override
+  public <T> T deserialize(MemoryBuffer buffer, TypeRef<T> typeRef) {
+    PooledEntry entry = acquire();
+    try {
+      return entry.fory.deserialize(buffer, typeRef);
+    } finally {
+      release(entry);
+    }
+  }
+
+  @Override
+  public <T> T deserialize(ForyInputStream inputStream, TypeRef<T> typeRef) {
+    PooledEntry entry = acquire();
+    try {
+      return entry.fory.deserialize(inputStream, typeRef);
+    } finally {
+      release(entry);
+    }
+  }
+
+  @Override
+  public <T> T deserialize(ForyReadableChannel channel, TypeRef<T> typeRef) {
+    PooledEntry entry = acquire();
+    try {
+      return entry.fory.deserialize(channel, typeRef);
+    } finally {
+      release(entry);
+    }
+  }
+
   @Override
   public Object deserialize(byte[] bytes, Iterable<MemoryBuffer> outOfBandBuffers) {
     PooledEntry entry = acquire();

java/fory-core/src/test/java/org/apache/fory/ForyTest.java

@@ -27,11 +27,13 @@
 import com.google.common.collect.HashBasedTable;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import java.io.ByteArrayInputStream;
 import java.io.Serializable;
 import java.lang.invoke.MethodHandles;
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
 import java.sql.Timestamp;
 import java.time.Instant;
 import java.time.LocalDate;
@@ -45,6 +47,8 @@
 import java.util.LinkedHashMap;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.StringTokenizer;
 import java.util.TreeMap;
 import java.util.TreeSet;
@@ -63,9 +67,12 @@
 import org.apache.fory.exception.ForyException;
 import org.apache.fory.exception.InsecureException;
 import org.apache.fory.exception.SerializationException;
+import org.apache.fory.io.ForyInputStream;
+import org.apache.fory.io.ForyReadableChannel;
 import org.apache.fory.memory.MemoryBuffer;
 import org.apache.fory.memory.MemoryUtils;
 import org.apache.fory.reflect.ReflectionUtils;
+import org.apache.fory.reflect.TypeRef;
 import org.apache.fory.resolver.TypeResolver;
 import org.apache.fory.serializer.ArraySerializersTest;
 import org.apache.fory.serializer.EnumSerializerTest;
@@ -980,4 +987,81 @@ public void testMaxDepthCodegen() {
             .build();
     assertThrows(InsecureException.class, () -> fory.deserialize(bytes));
   }
+
+  @Test
+  public void testDeserializeWithTypeRef() throws Exception {
+    Fory fory =
+        Fory.builder()
+            .requireClassRegistration(false)
+            .withXlang(false)
+            .withCompatible(false)
+            .build();
+
+    // 1. Root ArrayList<String> (exercises CollectionSerializer / CollectionLikeSerializer)
+    List<String> arrayList = new ArrayList<>(Arrays.asList("hello", "fory", "world"));
+    byte[] listBytes = fory.serialize(arrayList);
+    TypeRef<List<String>> listTypeRef = new TypeRef<List<String>>() {};
+
+    assertEquals(fory.deserialize(listBytes, listTypeRef), arrayList);
+    assertEquals(fory.deserialize(MemoryBuffer.fromByteArray(listBytes), listTypeRef), arrayList);
+
+    try (ForyInputStream stream = new ForyInputStream(new ByteArrayInputStream(listBytes))) {
+      assertEquals(fory.deserialize(stream, listTypeRef), arrayList);
+    }
+    try (ForyReadableChannel channel =
+        new ForyReadableChannel(Channels.newChannel(new ByteArrayInputStream(listBytes)))) {
+      assertEquals(fory.deserialize(channel, listTypeRef), arrayList);
+    }
+
+    // 2. Root LinkedList<String> and HashSet<String>
+    List<String> linkedList = new LinkedList<>(Arrays.asList("a", "b", "c"));
+    assertEquals(
+        fory.deserialize(fory.serialize(linkedList), new TypeRef<List<String>>() {}), linkedList);
+
+    Set<String> set = new HashSet<>(Arrays.asList("x", "y", "z"));
+    assertEquals(fory.deserialize(fory.serialize(set), new TypeRef<Set<String>>() {}), set);
+
+    // 3. Root Map<String, Integer> and nested Map<String, List<Integer>>
+    Map<String, Integer> intMap = new HashMap<>();
+    intMap.put("one", 1);
+    intMap.put("two", 2);
+    assertEquals(
+        fory.deserialize(fory.serialize(intMap), new TypeRef<Map<String, Integer>>() {}), intMap);
+
+    Map<String, List<Integer>> nestedMap = new HashMap<>();
+    nestedMap.put("key1", new ArrayList<>(Arrays.asList(1, 2, 3)));
+    nestedMap.put("key2", new ArrayList<>(Arrays.asList(4, 5, 6)));
+    byte[] mapBytes = fory.serialize(nestedMap);
+    TypeRef<Map<String, List<Integer>>> mapTypeRef = new TypeRef<Map<String, List<Integer>>>() {};
+    assertEquals(fory.deserialize(mapBytes, mapTypeRef), nestedMap);
+    assertEquals(fory.deserialize(MemoryBuffer.fromByteArray(mapBytes), mapTypeRef), nestedMap);
+
+    // 4. Arrays.asList view
+    List<String> asList = Arrays.asList("1", "2", "3");
+    assertEquals(fory.deserialize(fory.serialize(asList), new TypeRef<List<String>>() {}), asList);
+
+    // 5. Concrete object with TypeRef.of
+    BeanA beanA = BeanA.createBeanA(2);
+    byte[] beanBytes = fory.serialize(beanA);
+    assertEquals(fory.deserialize(beanBytes, TypeRef.of(BeanA.class)), beanA);
+
+    // 6. Null typeRef checks across all deserialize overloads
+    assertThrows(
+        NullPointerException.class,
+        () -> fory.deserialize(listBytes, (TypeRef<List<String>>) null));
+    assertThrows(
+        NullPointerException.class,
+        () ->
+            fory.deserialize(MemoryBuffer.fromByteArray(listBytes), (TypeRef<List<String>>) null));
+    try (ForyInputStream stream = new ForyInputStream(new ByteArrayInputStream(listBytes))) {
+      assertThrows(
+          NullPointerException.class, () -> fory.deserialize(stream, (TypeRef<List<String>>) null));
+    }
+    try (ForyReadableChannel channel =
+        new ForyReadableChannel(Channels.newChannel(new ByteArrayInputStream(listBytes)))) {
+      assertThrows(
+          NullPointerException.class,
+          () -> fory.deserialize(channel, (TypeRef<List<String>>) null));
+    }
+  }
 }

java/fory-core/src/test/java/org/apache/fory/StreamTest.java

@@ -44,6 +44,7 @@
 import org.apache.fory.io.ForyStreamReader;
 import org.apache.fory.memory.MemoryBuffer;
 import org.apache.fory.reflect.ReflectionUtils;
+import org.apache.fory.reflect.TypeRef;
 import org.apache.fory.test.bean.BeanA;
 import org.testng.Assert;
 import org.testng.annotations.Test;
@@ -272,6 +273,9 @@ public void testOutputStreamWithType() throws IOException {
     assertEquals(newObj, beanA);
     newObj = fory.deserialize(buf, BeanA.class);
     assertEquals(newObj, beanA);
+    buf.readerIndex(0);
+    newObj = fory.deserialize(buf, new TypeRef<BeanA>() {});
+    assertEquals(newObj, beanA);
   }
 
   @Test
@@ -311,6 +315,20 @@ public void testReadableChannel() throws IOException {
         Files.delete(tempFile);
       }
     }
+    {
+      ByteArrayOutputStream bas = new ByteArrayOutputStream();
+      fory.serialize(bas, beanA);
+
+      Path tempFile = Files.createTempFile("readable_channel_test", "data_3");
+      Files.write(tempFile, bas.toByteArray());
+
+      try (ForyReadableChannel channel = of(Files.newByteChannel(tempFile))) {
+        Object newObj = fory.deserialize(channel, new TypeRef<BeanA>() {});
+        assertEquals(newObj, beanA);
+      } finally {
+        Files.delete(tempFile);
+      }
+    }
   }
 
   @Test