Skip to content

feat: Align object array to collection serialization protocol v2 #2218

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 5 commits into
base: main
Choose a base branch
from
+135 −66
Draft

feat: Align object array to collection serialization protocol v2 #2218

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
561c780
feat: Align object array to collection serialization protocol v2
david1437 May 11, 2025
56002e2
fix: build collection type dynamically with TypeUtils
david1437 May 12, 2025
3cdf8ef
fix: cache ArrayAsList and use collectionm serializer for 1D objects …
david1437 May 12, 2025
7e2d793
fix: handle depth during write serialization
david1437 May 12, 2025
24ee77c
fix: Handle primitive casting in read
david1437 May 16, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
193 changes: 127 additions & 66 deletions java/fury-core/src/main/java/org/apache/fury/serializer/ArraySerializers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,12 +26,17 @@
import org.apache.fury.config.CompatibleMode;
import org.apache.fury.memory.MemoryBuffer;
import org.apache.fury.memory.Platform;
import org.apache.fury.reflect.TypeRef;
import org.apache.fury.resolver.ClassInfo;
import org.apache.fury.resolver.ClassInfoHolder;
import org.apache.fury.resolver.ClassResolver;
import org.apache.fury.resolver.RefResolver;
import org.apache.fury.serializer.NonexistentClass.NonexistentSkip;
import org.apache.fury.serializer.NonexistentClassSerializers.NonexistentEnumClassSerializer;
import org.apache.fury.serializer.Serializers.CrossLanguageCompatibleSerializer;
import org.apache.fury.serializer.collection.CollectionFlags;
import org.apache.fury.serializer.collection.FuryArrayAsListSerializer;
import org.apache.fury.serializer.collection.FuryArrayAsListSerializer.ArrayAsList;
import org.apache.fury.type.GenericType;
import org.apache.fury.type.TypeUtils;
import org.apache.fury.type.Types;
Expand All @@ -45,6 +50,9 @@ public class ArraySerializers {
public static final class ObjectArraySerializer<T> extends Serializer<T[]> {
private final Class<T> innerType;
private final Serializer componentTypeSerializer;
private final FuryArrayAsListSerializer collectionSerializer;
private ArrayAsList list;
private final GenericType collectionGenericType;
private final ClassInfoHolder classInfoHolder;
private final int[] stubDims;
private final GenericType componentGenericType;
Expand All @@ -69,14 +77,21 @@ public ObjectArraySerializer(Fury fury, Class<T[]> cls) {
if (fury.getClassResolver().isMonomorphic(componentType)) {
if (fury.isCrossLanguage()) {
this.componentTypeSerializer = null;
this.collectionSerializer = null;
this.collectionGenericType = null;
} else {
this.componentTypeSerializer = fury.getClassResolver().getSerializer(componentType);
this.collectionSerializer = new FuryArrayAsListSerializer(fury);
this.collectionGenericType = buildCollectionGenericType(dimension);
}
} else {
// TODO add ClassInfo cache for non-final component type.
this.componentTypeSerializer = null;
this.collectionSerializer = null;
this.collectionGenericType = null;
}
this.stubDims = new int[dimension];
this.list = null;
classInfoHolder = fury.getClassResolver().nilClassInfoHolder();
}

Expand All @@ -85,52 +100,81 @@ public void write(MemoryBuffer buffer, T[] arr) {
int len = arr.length;
RefResolver refResolver = fury.getRefResolver();
Serializer componentSerializer = this.componentTypeSerializer;
int header = componentSerializer != null ? 0b1 : 0b0;
buffer.writeVarUint32Small7(len << 1 | header);
if (componentSerializer != null) {
for (T t : arr) {
if (!refResolver.writeRefOrNull(buffer, t)) {
componentSerializer.write(buffer, t);
}
if (this.collectionSerializer != null) {
buffer.writeVarUint32Small7(len);
if (len == 0) {
return;
}
ArrayAsList list = this.list;
if (list == null) {
list = new ArrayAsList(0);
} else {
this.list = null;
}
list.setArray(arr);
fury.incDepth(1);
fury.getGenerics().pushGenericType(this.collectionGenericType);
this.collectionSerializer.write(buffer, list);
fury.getGenerics().popGenericType();
fury.incDepth(-1);
this.list = list;
} else {
Fury fury = this.fury;
ClassResolver classResolver = fury.getClassResolver();
ClassInfo classInfo = null;
Class<?> elemClass = null;
for (T t : arr) {
if (!refResolver.writeRefOrNull(buffer, t)) {
Class<?> clz = t.getClass();
if (clz != elemClass) {
elemClass = clz;
classInfo = classResolver.getClassInfo(clz);
int header = componentSerializer != null ? 0b1 : 0b0;
buffer.writeVarUint32Small7(len << 1 | header);
if (componentSerializer != null) {
for (T t : arr) {
if (!refResolver.writeRefOrNull(buffer, t)) {
componentSerializer.write(buffer, t);
}
}
} else {
Fury fury = this.fury;
ClassResolver classResolver = fury.getClassResolver();
ClassInfo classInfo = null;
Class<?> elemClass = null;
for (T t : arr) {
if (!refResolver.writeRefOrNull(buffer, t)) {
Class<?> clz = t.getClass();
if (clz != elemClass) {
elemClass = clz;
classInfo = classResolver.getClassInfo(clz);
}
fury.writeNonRef(buffer, t, classInfo);
}
fury.writeNonRef(buffer, t, classInfo);
}
}
}
}

@Override
public T[] copy(T[] originArray) {
int length = originArray.length;
Object[] newArray = newArray(length);
if (needToCopyRef) {
fury.reference(originArray, newArray);
}
Object[] newArray;
Serializer componentSerializer = this.componentTypeSerializer;
if (componentSerializer != null) {
if (componentSerializer.isImmutable()) {
System.arraycopy(originArray, 0, newArray, 0, length);
if (this.collectionSerializer != null) {
fury.getGenerics().pushGenericType(this.collectionGenericType);
ArrayAsList list = new ArrayAsList(originArray.length);
list.setArray(originArray);
newArray = this.collectionSerializer.copy(list).toArray();
fury.getGenerics().popGenericType();
} else {
int length = originArray.length;
newArray = newArray(length);
if (needToCopyRef) {
fury.reference(originArray, newArray);
}
if (componentSerializer != null) {
if (componentSerializer.isImmutable()) {
System.arraycopy(originArray, 0, newArray, 0, length);
} else {
for (int i = 0; i < length; i++) {
newArray[i] = componentSerializer.copy(originArray[i]);
}
}
} else {
for (int i = 0; i < length; i++) {
newArray[i] = componentSerializer.copy(originArray[i]);
newArray[i] = fury.copyObject(originArray[i]);
}
}
} else {
for (int i = 0; i < length; i++) {
newArray[i] = fury.copyObject(originArray[i]);
}
}
return (T[]) newArray;
}
Expand All @@ -147,39 +191,49 @@ public void xwrite(MemoryBuffer buffer, T[] arr) {

@Override
public T[] read(MemoryBuffer buffer) {
Object[] value;
int numElements = buffer.readVarUint32Small7();
boolean isFinal = (numElements & 0b1) != 0;
numElements >>>= 1;
Object[] value = newArray(numElements);
RefResolver refResolver = fury.getRefResolver();
refResolver.reference(value);
if (isFinal) {
final Serializer componentTypeSerializer = this.componentTypeSerializer;
for (int i = 0; i < numElements; i++) {
Object elem;
int nextReadRefId = refResolver.tryPreserveRefId(buffer);
if (nextReadRefId >= Fury.NOT_NULL_VALUE_FLAG) {
elem = componentTypeSerializer.read(buffer);
refResolver.setReadObject(nextReadRefId, elem);
} else {
elem = refResolver.getReadObject();
}
value[i] = elem;
if (this.collectionSerializer != null) {
fury.getGenerics().pushGenericType(this.collectionGenericType);
value = this.collectionSerializer.read(buffer).toArray();
fury.getGenerics().popGenericType();
if (this.innerType.isPrimitive() || TypeUtils.isBoxed(this.innerType)) {
return Arrays.copyOf(value, value.length, this.type);
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Considering this does a copy it does not feel good is there a better approach for handling this? without this i get a class cast exception?

}
} else {
Fury fury = this.fury;
ClassInfoHolder classInfoHolder = this.classInfoHolder;
for (int i = 0; i < numElements; i++) {
int nextReadRefId = refResolver.tryPreserveRefId(buffer);
Object o;
if (nextReadRefId >= Fury.NOT_NULL_VALUE_FLAG) {
// ref value or not-null value
o = fury.readNonRef(buffer, classInfoHolder);
refResolver.setReadObject(nextReadRefId, o);
} else {
o = refResolver.getReadObject();
boolean isFinal = (numElements & 0b1) != 0;
numElements >>>= 1;
value = newArray(numElements);
RefResolver refResolver = fury.getRefResolver();
refResolver.reference(value);
if (isFinal) {
final Serializer componentTypeSerializer = this.componentTypeSerializer;
for (int i = 0; i < numElements; i++) {
Object elem;
int nextReadRefId = refResolver.tryPreserveRefId(buffer);
if (nextReadRefId >= Fury.NOT_NULL_VALUE_FLAG) {
elem = componentTypeSerializer.read(buffer);
refResolver.setReadObject(nextReadRefId, elem);
} else {
elem = refResolver.getReadObject();
}
value[i] = elem;
}
} else {
Fury fury = this.fury;
ClassInfoHolder classInfoHolder = this.classInfoHolder;
for (int i = 0; i < numElements; i++) {
int nextReadRefId = refResolver.tryPreserveRefId(buffer);
Object o;
if (nextReadRefId >= Fury.NOT_NULL_VALUE_FLAG) {
// ref value or not-null value
o = fury.readNonRef(buffer, classInfoHolder);
refResolver.setReadObject(nextReadRefId, o);
} else {
o = refResolver.getReadObject();
}
value[i] = o;
}
value[i] = o;
}
}
return (T[]) value;
Expand Down Expand Up @@ -208,6 +262,14 @@ private Object[] newArray(int numElements) {
}
return value;
}

private GenericType buildCollectionGenericType(int dims) {
TypeRef arrayType = TypeRef.of(this.innerType);
for (int i = 0; i < dims; i++) {
arrayType = TypeUtils.collectionOf(arrayType);
}
return GenericType.build(arrayType);
}
}

public static final class PrimitiveArrayBufferObject implements BufferObject {
Expand Down Expand Up @@ -249,7 +311,7 @@ public MemoryBuffer toBuffer() {
// Implement all read/write methods in subclasses to avoid
// virtual method call cost.
public abstract static class PrimitiveArraySerializer<T>
extends Serializers.CrossLanguageCompatibleSerializer<T> {
extends CrossLanguageCompatibleSerializer<T> {
protected final int offset;
protected final int elemSize;

Expand Down Expand Up @@ -610,14 +672,14 @@ public double[] read(MemoryBuffer buffer) {
public static final class StringArraySerializer extends Serializer<String[]> {
private final StringSerializer stringSerializer;
private final FuryArrayAsListSerializer collectionSerializer;
private final FuryArrayAsListSerializer.ArrayAsList list;
private final ArrayAsList list;

public StringArraySerializer(Fury fury) {
super(fury, String[].class);
stringSerializer = new StringSerializer(fury);
collectionSerializer = new FuryArrayAsListSerializer(fury);
collectionSerializer.setElementSerializer(stringSerializer);
list = new FuryArrayAsListSerializer.ArrayAsList(0);
list = new ArrayAsList(0);
}

@Override
Expand Down Expand Up @@ -890,11 +952,10 @@ public NonexistentArrayClassSerializer(Fury fury, Class<?> cls) {
public NonexistentArrayClassSerializer(Fury fury, String className, Class<?> cls) {
super(fury, className, cls);
if (TypeUtils.getArrayComponent(cls).isEnum()) {
componentSerializer = new NonexistentClassSerializers.NonexistentEnumClassSerializer(fury);
componentSerializer = new NonexistentEnumClassSerializer(fury);
} else {
if (fury.getConfig().getCompatibleMode() == CompatibleMode.COMPATIBLE) {
componentSerializer =
new CompatibleSerializer<>(fury, NonexistentClass.NonexistentSkip.class);
componentSerializer = new CompatibleSerializer<>(fury, NonexistentSkip.class);
} else {
componentSerializer = null;
}
Expand Down
7 changes: 7 additions & 0 deletions ...y-core/src/main/java/org/apache/fury/serializer/collection/FuryArrayAsListSerializer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,13 @@ public Collection newCollection(MemoryBuffer buffer) {
return new ArrayAsList(numElements);
}

@Override
public Collection newCollection(Collection collection) {
int numElements = collection.size();
setNumElements(numElements);
return new ArrayAsList(numElements);
}

/**
* A List which wrap a Java array into a list, used for serialization only, do not use it in other
* scenarios.
Expand Down
1 change: 1 addition & 0 deletions java/fury-core/src/test/java/org/apache/fury/serializer/ArraySerializersTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,7 @@ public void testMultiArraySerialization(boolean referenceTracking, Language lang
{false, true, (byte) 1, (byte) 1, (float) 1.0, (float) 1.1}
});
serDeCheckTyped(fury1, fury2, new Integer[][] {{1, 2}, {1, 2}});
serDeCheckTyped(fury1, fury2, new String[][][] {{{"str", "str"}, {"str", "str"}}});
}

@Test(dataProvider = "furyCopyConfig")
Expand Down
Loading