You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fory.apache.org by ch...@apache.org on 2025/10/14 02:45:45 UTC
(fory) branch main updated: feat(java): introduce Compact Row Codec (#2414)
This is an automated email from the ASF dual-hosted git repository.
chaokunyang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fory.git
The following commit(s) were added to refs/heads/main by this push:
new ee3c29426 feat(java): introduce Compact Row Codec (#2414)
ee3c29426 is described below
commit ee3c29426181ac19b75614a68075436bf439e6f0
Author: Steven Schlansker <st...@apache.org>
AuthorDate: Mon Oct 13 19:45:42 2025 -0700
feat(java): introduce Compact Row Codec (#2414)
## What does this PR do?
Introduce alternate "compact" row encoding that better uses knowledge of
fixed-size types and sacrifices alignment to save space.
Introduce new Builder pattern to avoid explosion of `Encoders` static
methods as more features are added to row format.
Optimizations include:
* struct stores fixed-size fields (e.g. Int128. FixedSizeBinary) inline
in fixed-data area without offset + size
* struct of all fixed-sized fields is itself considered fixed-size to
store in other struct or array
* struct skips null bitmap if all fields are non-nullable
* struct sorts fields by fixed-size for best-effort (but not guaranteed)
alignment
* struct can use less than 8 bytes for small data (int, short, etc)
* struct null bitmap stored at end of struct to borrow alignment padding
if possible
* array stores fixed-size fields inline in fixed-data area without
offset+size
* array header uses 4 bytes for size (since Collection and array are
only int-sized) and leaves remaining 4 bytes for start of null bitmap
Fixups include:
* toString better handles varbinary / fixed-binary (hex dump of first
256 bytes)
* start making Javadoc for row format
Compromises:
* less alignment could increase access time, but this is opt-in. and I
think on modern processors it is not such a big deal.
* increased complexity of offset lookup, try to pre-compute in an array
when possible and use `StableValue` when it is GA
Not compatible with existing row format.
## Related issues
Fixes #2337
## Does this PR introduce any user-facing change?
New API for new Compact codec. Existing codec unchanged.
---
.../java/org/apache/fory/builder/CodecBuilder.java | 2 +
.../java/org/apache/fory/memory/MemoryBuffer.java | 2 +-
.../java/org/apache/fory/util/ExceptionUtils.java | 9 +
java/fory-format/README.md | 55 +-
.../fory/format/encoder/ArrayCodecBuilder.java | 122 ++++
.../fory/format/encoder/ArrayEncoderBuilder.java | 9 +-
.../format/encoder/BaseBinaryEncoderBuilder.java | 144 +++--
.../fory/format/encoder/BaseCodecBuilder.java | 75 +++
.../fory/format/encoder/BinaryArrayEncoder.java | 101 +++
.../fory/format/encoder/BinaryMapEncoder.java | 118 ++++
.../fory/format/encoder/BinaryRowEncoder.java | 125 ++++
.../encoder/BufferResettingArrayEncoder.java | 80 +++
.../format/encoder/BufferResettingMapEncoder.java | 93 +++
.../format/encoder/BufferResettingRowEncoder.java | 78 +++
.../format/encoder/CompactArrayEncoderBuilder.java | 61 ++
.../fory/format/encoder/CompactCodecFormat.java | 93 +++
.../format/encoder/CompactMapEncoderBuilder.java | 69 ++
.../format/encoder/CompactRowEncoderBuilder.java | 127 ++++
.../fory/format/encoder/DefaultCodecFormat.java | 89 +++
.../org/apache/fory/format/encoder/Encoder.java | 14 +-
.../org/apache/fory/format/encoder/Encoders.java | 515 ++-------------
.../org/apache/fory/format/encoder/Encoding.java | 55 ++
.../fory/format/encoder/GeneratedRowEncoder.java | 3 +-
.../fory/format/encoder/MapCodecBuilder.java | 108 ++++
.../fory/format/encoder/MapEncoderBuilder.java | 25 +-
.../fory/format/encoder/RowCodecBuilder.java | 98 +++
.../org/apache/fory/format/encoder/RowEncoder.java | 5 +-
.../fory/format/encoder/RowEncoderBuilder.java | 28 +-
.../java/org/apache/fory/format/row/Getters.java | 2 +-
.../fory/format/row/ToStringValueVisitor.java | 57 ++
.../org/apache/fory/format/row/ValueVisitor.java | 5 +
.../apache/fory/format/row/binary/BinaryArray.java | 50 +-
.../apache/fory/format/row/binary/BinaryMap.java | 18 +-
.../apache/fory/format/row/binary/BinaryRow.java | 39 +-
.../fory/format/row/binary/CompactBinaryArray.java | 138 ++++
.../binary/CompactBinaryMap.java} | 32 +-
.../fory/format/row/binary/CompactBinaryRow.java | 157 +++++
.../apache/fory/format/row/binary/UnsafeTrait.java | 37 +-
...naryRowWriter.java => BaseBinaryRowWriter.java} | 99 ++-
.../row/binary/writer/BinaryArrayWriter.java | 69 +-
.../format/row/binary/writer/BinaryRowWriter.java | 78 +--
.../format/row/binary/writer/BinaryWriter.java | 52 +-
.../binary/writer/CompactBinaryArrayWriter.java | 147 +++++
.../row/binary/writer/CompactBinaryRowWriter.java | 336 ++++++++++
.../binary/writer/FieldAlignmentComparator.java} | 36 +-
.../org/apache/fory/format/type/DataTypes.java | 14 +-
.../org/apache/fory/format/type/TypeInference.java | 4 +
.../fory/format/encoder/ArrayEncoderTest.java | 2 +-
.../fory/format/encoder/CodecBuilderTest.java | 7 +-
.../fory/format/encoder/CompactCodecTest.java | 708 +++++++++++++++++++++
.../fory/format/encoder/CustomCodecTest.java | 7 +-
.../apache/fory/format/encoder/RowEncoderTest.java | 3 +-
52 files changed, 3632 insertions(+), 768 deletions(-)
diff --git a/java/fory-core/src/main/java/org/apache/fory/builder/CodecBuilder.java b/java/fory-core/src/main/java/org/apache/fory/builder/CodecBuilder.java
index 4c3f2498d..99f827b07 100644
--- a/java/fory-core/src/main/java/org/apache/fory/builder/CodecBuilder.java
+++ b/java/fory-core/src/main/java/org/apache/fory/builder/CodecBuilder.java
@@ -124,6 +124,8 @@ public abstract class CodecBuilder {
.forEach(ctx::reserveName);
}
+ public abstract String codecClassName(Class<?> cls);
+
/** Generate codec class code. */
public abstract String genCode();
diff --git a/java/fory-core/src/main/java/org/apache/fory/memory/MemoryBuffer.java b/java/fory-core/src/main/java/org/apache/fory/memory/MemoryBuffer.java
index a444e48a9..29331f269 100644
--- a/java/fory-core/src/main/java/org/apache/fory/memory/MemoryBuffer.java
+++ b/java/fory-core/src/main/java/org/apache/fory/memory/MemoryBuffer.java
@@ -40,7 +40,7 @@ import sun.misc.Unsafe;
* <li>additional binary compare, swap, and copy methods.
* <li>little-endian access.
* <li>independent read/write index.
- * <li>variant int/long encoding.
+ * <li>varint int/long encoding.
* <li>aligned int/long encoding.
* </ul>
*
diff --git a/java/fory-core/src/main/java/org/apache/fory/util/ExceptionUtils.java b/java/fory-core/src/main/java/org/apache/fory/util/ExceptionUtils.java
index 5e08141dc..e4577f9b6 100644
--- a/java/fory-core/src/main/java/org/apache/fory/util/ExceptionUtils.java
+++ b/java/fory-core/src/main/java/org/apache/fory/util/ExceptionUtils.java
@@ -71,4 +71,13 @@ public class ExceptionUtils {
}
public static void ignore(Object... args) {}
+
+ public static RuntimeException throwAnyway(Throwable t) {
+ throw ExceptionUtils.<RuntimeException>throwEvadingChecks(t);
+ }
+
+ @SuppressWarnings({"unchecked", "TypeParameterUnusedInFormals"})
+ private static <E extends Throwable> E throwEvadingChecks(Throwable throwable) throws E {
+ throw (E) throwable;
+ }
}
diff --git a/java/fory-format/README.md b/java/fory-format/README.md
index e867e11cf..20b8fadcf 100644
--- a/java/fory-format/README.md
+++ b/java/fory-format/README.md
@@ -8,12 +8,61 @@ Fory row format is heavily inspired by spark tungsten row format, but with chang
- Decimal use arrow decimal format.
- Variable-size field can be inline in fixed-size region if small enough.
- Allow skip padding by generate Row using aot to put offsets in generated code.
-- Support adding fields without breaking compatibility.
-The initial fory java row data structure implementation is modified from spark unsafe row/writer.
+The initial Fory java row data structure implementation is modified from spark unsafe row/writer.
See `Encoders.bean` Javadoc for a list built-in supported types.
+## Row Format Java
+
+To begin using the row format from Java, start with the `Encoders` class:
+
+```
+// Many built-in types and collections are supported
+public record MyRecord(int key, String value) {}
+
+// The encoder supplier is relatively expensive to create
+// It is thread-safe and should be re-used
+Supplier<RowEncoder<MyRecord>> encoderFactory =
+ Encoders.buildBeanCodec(MyRecord.class)
+ .build();
+
+// Each individual encoder is relatively cheap to create
+// It is not thread-safe, but may be reused by the same thread
+var encoder = encoderFactory.get();
+byte[] encoded = encoder.encode(new MyRecord(42, "Test"));
+
+MyRecord deserialized = encoder.decode(encoded);
+```
+
+## Compact Format
+
+The default row format is cross-language compatible and alignment-padded for maximum performance.
+When data size is a greater concern, the compact format provides an alternate encoding that uses
+significantly less space.
+
+Enable the compact codec on the encoder builder:
+
+```
+Supplier<RowEncoder<MyRecord>> encoderFactory =
+ Encoders.buildBeanCodec(MyRecord.class)
+ .compactEncoding()
+ .build();
+```
+
+Optimizations include:
+
+- struct stores fixed-size fields (e.g. Int128. FixedSizeBinary) inline in fixed-data area without offset + size
+- struct of all fixed-sized fields is itself considered fixed-size to store in other struct or array
+- struct skips null bitmap if all fields are non-nullable
+- struct sorts fields by fixed-size for best-effort (but not guaranteed) alignment
+- struct can use less than 8 bytes for small data (int, short, etc)
+- struct null bitmap stored at end of struct to borrow alignment padding if possible
+- array stores fixed-size fields inline in fixed-data area without offset+size
+- array header uses 4 bytes for size (since Collection and array are only int-sized) and leaves remaining 4 bytes for start of null bitmap
+
+## Custom Type Registration
+
It is possible to register custom type handling and collection factories for the row format -
see Encoders.registerCustomCodec and Encoders.registerCustomCollectionFactory. For an interface,
Fory can synthesize a simple value implementation, such as the UuidType below.
@@ -45,7 +94,7 @@ static class UuidEncoder implements CustomCodec.MemoryBufferCodec<UUID> {
static class SortedSetOfUuidDecoder implements CustomCollectionFactory<UUID, SortedSet<UUID>> {
@Override
public SortedSet<UUID> newCollection(final int size) {
- return new TreeSet<>(UnsignedUuidComparator.INSTANCE);
+ return new TreeSet<>();
}
}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/ArrayCodecBuilder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/ArrayCodecBuilder.java
new file mode 100644
index 000000000..a2faa102a
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/ArrayCodecBuilder.java
@@ -0,0 +1,122 @@
+/*
+ * 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.fory.format.encoder;
+
+import static org.apache.fory.type.TypeUtils.getRawType;
+
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
+import org.apache.fory.format.type.DataTypes;
+import org.apache.fory.format.type.TypeInference;
+import org.apache.fory.reflect.TypeRef;
+import org.apache.fory.type.TypeUtils;
+import org.apache.fory.util.ExceptionUtils;
+
+public class ArrayCodecBuilder<C extends Collection<?>>
+ extends BaseCodecBuilder<ArrayCodecBuilder<C>> {
+
+ private final TypeRef<C> collectionType;
+ private final Field elementField;
+
+ ArrayCodecBuilder(final TypeRef<C> collectionType) {
+ super(TypeInference.inferSchema(collectionType, false));
+ this.collectionType = collectionType;
+ elementField = DataTypes.fieldOfSchema(schema, 0);
+ }
+
+ public Supplier<ArrayEncoder<C>> build() {
+ final Function<BinaryArrayWriter, ArrayEncoder<C>> arrayEncoderFactory = buildWithWriter();
+ return new Supplier<ArrayEncoder<C>>() {
+ @Override
+ public ArrayEncoder<C> get() {
+ final BinaryArrayWriter writer = codecFormat.newArrayWriter(elementField);
+ return new BufferResettingArrayEncoder<>(
+ initialBufferSize, writer, arrayEncoderFactory.apply(writer));
+ }
+ };
+ }
+
+ Function<BinaryArrayWriter, ArrayEncoder<C>> buildWithWriter() {
+ loadArrayInnerCodecs();
+ final Function<BinaryArrayWriter, GeneratedArrayEncoder> generatedEncoderFactory =
+ generatedEncoderFactory();
+ return new Function<BinaryArrayWriter, ArrayEncoder<C>>() {
+ @Override
+ public ArrayEncoder<C> apply(final BinaryArrayWriter writer) {
+ return new BinaryArrayEncoder<>(
+ writer, generatedEncoderFactory.apply(writer), sizeEmbedded);
+ }
+ };
+ }
+
+ private void loadArrayInnerCodecs() {
+ final Set<TypeRef<?>> set = new HashSet<>();
+ Encoders.findBeanToken(collectionType, set);
+ if (set.isEmpty()) {
+ throw new IllegalArgumentException("can not find bean class.");
+ }
+
+ for (final TypeRef<?> tt : set) {
+ Encoders.loadOrGenRowCodecClass(getRawType(tt), codecFormat);
+ }
+ }
+
+ Function<BinaryArrayWriter, GeneratedArrayEncoder> generatedEncoderFactory() {
+ final TypeRef<?> elementType = TypeUtils.getElementType(collectionType);
+ final Class<?> arrayCodecClass =
+ Encoders.loadOrGenArrayCodecClass(collectionType, elementType, codecFormat);
+
+ final MethodHandle constructorHandle;
+ try {
+ final var constructor =
+ arrayCodecClass.asSubclass(GeneratedArrayEncoder.class).getConstructor(Object[].class);
+ constructorHandle =
+ MethodHandles.lookup()
+ .unreflectConstructor(constructor)
+ .asType(MethodType.methodType(GeneratedArrayEncoder.class, Object[].class));
+ } catch (final NoSuchMethodException | IllegalAccessException e) {
+ throw new EncoderException(
+ "Failed to construct array codec for "
+ + collectionType
+ + " with element class "
+ + elementType,
+ e);
+ }
+ return new Function<BinaryArrayWriter, GeneratedArrayEncoder>() {
+ @Override
+ public GeneratedArrayEncoder apply(final BinaryArrayWriter writer) {
+ final Object[] references = {writer.getField(), writer, fory};
+ try {
+ return (GeneratedArrayEncoder) constructorHandle.invokeExact(references);
+ } catch (final Throwable t) {
+ throw ExceptionUtils.throwAnyway(t);
+ }
+ }
+ };
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/ArrayEncoderBuilder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/ArrayEncoderBuilder.java
index 2865d62c8..f95473da7 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/encoder/ArrayEncoderBuilder.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/ArrayEncoderBuilder.java
@@ -31,7 +31,6 @@ import org.apache.fory.codegen.CodegenContext;
import org.apache.fory.codegen.Expression;
import org.apache.fory.codegen.ExpressionUtils;
import org.apache.fory.format.row.binary.BinaryArray;
-import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
import org.apache.fory.format.type.TypeInference;
import org.apache.fory.logging.Logger;
import org.apache.fory.logging.LoggerFactory;
@@ -104,13 +103,13 @@ public class ArrayEncoderBuilder extends BaseBinaryEncoderBuilder {
"arrayWriter",
ROOT_ARRAY_WRITER_NAME,
"arrayWriterType",
- ctx.type(BinaryArrayWriter.class),
+ arrayWriterType(),
"fory",
FORY_NAME,
"foryType",
ctx.type(Fory.class));
ctx.addField(ctx.type(Field.class), FIELD_NAME);
- ctx.addField(ctx.type(BinaryArrayWriter.class), ROOT_ARRAY_WRITER_NAME);
+ ctx.addField(ctx.type(arrayWriterType()), ROOT_ARRAY_WRITER_NAME);
ctx.addField(ctx.type(Fory.class), FORY_NAME);
Expression encodeExpr = buildEncodeExpression();
@@ -136,7 +135,7 @@ public class ArrayEncoderBuilder extends BaseBinaryEncoderBuilder {
@Override
public Expression buildEncodeExpression() {
Expression.Reference arrayWriter =
- new Expression.Reference(ROOT_ARRAY_WRITER_NAME, arrayWriterTypeToken, false);
+ new Expression.Reference(ROOT_ARRAY_WRITER_NAME, arrayWriterType(), false);
Expression.ListExpression expressions = new Expression.ListExpression();
Expression.Reference inputObject =
@@ -148,7 +147,7 @@ public class ArrayEncoderBuilder extends BaseBinaryEncoderBuilder {
Expression.Reference fieldExpr = new Expression.Reference(FIELD_NAME, ARROW_FIELD_TYPE, false);
Expression listExpression =
- serializeForArrayByWriter(array, arrayWriter, arrayToken, fieldExpr);
+ serializeForArrayByWriter(array, arrayWriter, arrayToken, null, fieldExpr);
expressions.add(listExpression);
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/BaseBinaryEncoderBuilder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BaseBinaryEncoderBuilder.java
index 102a76f7e..d20892121 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/encoder/BaseBinaryEncoderBuilder.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BaseBinaryEncoderBuilder.java
@@ -64,6 +64,7 @@ import org.apache.fory.format.row.Row;
import org.apache.fory.format.row.binary.BinaryArray;
import org.apache.fory.format.row.binary.BinaryRow;
import org.apache.fory.format.row.binary.BinaryUtils;
+import org.apache.fory.format.row.binary.writer.BaseBinaryRowWriter;
import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
import org.apache.fory.format.row.binary.writer.BinaryRowWriter;
import org.apache.fory.format.row.binary.writer.BinaryWriter;
@@ -86,9 +87,6 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
protected static final TypeRef<Field> ARROW_FIELD_TYPE = TypeRef.of(Field.class);
protected static TypeRef<Schema> schemaTypeToken = TypeRef.of(Schema.class);
protected static TypeRef<BinaryWriter> writerTypeToken = TypeRef.of(BinaryWriter.class);
- protected static TypeRef<BinaryRowWriter> rowWriterTypeToken = TypeRef.of(BinaryRowWriter.class);
- protected static TypeRef<BinaryArrayWriter> arrayWriterTypeToken =
- TypeRef.of(BinaryArrayWriter.class);
protected static TypeRef<Row> rowTypeToken = TypeRef.of(Row.class);
protected static TypeRef<BinaryRow> binaryRowTypeToken = TypeRef.of(BinaryRow.class);
protected static TypeRef<BinaryArray> binaryArrayTypeToken = TypeRef.of(BinaryArray.class);
@@ -120,6 +118,7 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
this.typeCtx = typeCtx;
}
+ @Override
public String codecClassName(Class<?> beanClass) {
return codecClassName(beanClass, "");
}
@@ -145,6 +144,14 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
return CodeGenerator.getPackage(beanClass) + "." + codecClassName(beanClass, prefix);
}
+ protected TypeRef<? extends BaseBinaryRowWriter> rowWriterType() {
+ return TypeRef.of(BinaryRowWriter.class);
+ }
+
+ protected TypeRef<? extends BinaryArrayWriter> arrayWriterType() {
+ return TypeRef.of(BinaryArrayWriter.class);
+ }
+
/**
* Return an expression for serializing an object of given type to row format representation. The
* inputObject will be written to position <code>ordinal</code> of row/array using given <code>
@@ -155,6 +162,7 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
Expression inputObject,
Expression writer,
TypeRef<?> typeRef,
+ Field fieldIfKnown,
Expression arrowField,
Set<TypeRef<?>> visitedCustomTypes) {
Class<?> rawType = getRawType(typeRef);
@@ -166,7 +174,13 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
visitedCustomTypes.add(typeRef);
Expression doSerialize =
serializeFor(
- ordinal, newInputObject, writer, rewrittenType, arrowField, visitedCustomTypes);
+ ordinal,
+ newInputObject,
+ writer,
+ rewrittenType,
+ fieldIfKnown,
+ arrowField,
+ visitedCustomTypes);
return new If(
new Expression.IsNull(inputObject),
new Invoke(writer, "setNullAt", ordinal),
@@ -188,6 +202,7 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
new Expression.Cast(unwrapped, elemType),
writer,
elemType,
+ fieldIfKnown,
arrowField,
visitedCustomTypes);
} else if (TypeUtils.isPrimitive(rawType)) {
@@ -254,24 +269,27 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
// place outer writer operations here, because map key/value arrays need to call
// serializeForArray,
// but don't setOffsetAndSize for array.
- Invoke offset =
+ Invoke originalWriterIndex =
new Invoke(writer, "writerIndex", "writerIndex", TypeUtils.PRIMITIVE_INT_TYPE);
- Expression serializeArray = serializeForArray(inputObject, writer, typeRef, arrowField);
+ Expression serializeArray =
+ serializeForArray(inputObject, writer, typeRef, fieldIfKnown, arrowField);
Arithmetic size =
ExpressionUtils.subtract(
new Invoke(writer, "writerIndex", "writerIndex", TypeUtils.PRIMITIVE_INT_TYPE),
- offset);
- Invoke setOffsetAndSize = new Invoke(writer, "setOffsetAndSize", ordinal, offset, size);
+ originalWriterIndex);
+ Invoke setOffsetAndSize =
+ new Invoke(writer, "setOffsetAndSize", ordinal, originalWriterIndex, size);
+ Expression finishArrayWrite = new ListExpression(size, setOffsetAndSize);
ListExpression expression =
- new ListExpression(offset, serializeArray, size, setOffsetAndSize);
+ new ListExpression(originalWriterIndex, serializeArray, finishArrayWrite);
return new If(
ExpressionUtils.eqNull(inputObject),
new Invoke(writer, "setNullAt", ordinal),
expression);
} else if (TypeUtils.MAP_TYPE.isSupertypeOf(typeRef)) {
- return serializeForMap(ordinal, writer, inputObject, typeRef, arrowField);
+ return serializeForMap(ordinal, writer, inputObject, typeRef, fieldIfKnown, arrowField);
} else if (TypeUtils.isBean(rawType, typeCtx)) {
- return serializeForBean(ordinal, writer, inputObject, typeRef, arrowField);
+ return serializeForBean(ordinal, writer, inputObject, fieldIfKnown, typeRef, arrowField);
} else if (rawType == BinaryArray.class) {
Invoke writeExp =
new Invoke(
@@ -302,20 +320,28 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
}
protected Expression serializeForArray(
- Expression inputObject, Expression writer, TypeRef<?> typeRef, Expression arrowField) {
+ Expression inputObject,
+ Expression writer,
+ TypeRef<?> typeRef,
+ Field fieldIfKnown,
+ Expression arrowField) {
Reference arrayWriter = getOrCreateArrayWriter(typeRef, arrowField, writer);
- return serializeForArrayByWriter(inputObject, arrayWriter, typeRef, arrowField);
+ return serializeForArrayByWriter(inputObject, arrayWriter, typeRef, fieldIfKnown, arrowField);
}
protected Expression serializeForArrayByWriter(
- Expression inputObject, Expression arrayWriter, TypeRef<?> typeRef, Expression arrowField) {
+ Expression inputObject,
+ Expression arrayWriter,
+ TypeRef<?> typeRef,
+ Field fieldIfKnown,
+ Expression arrowField) {
StaticInvoke arrayElementField =
new StaticInvoke(
DataTypes.class, "arrayElementField", "elemField", ARROW_FIELD_TYPE, false, arrowField);
Class<?> rawType = getRawType(typeRef);
if (rawType.isArray()) {
FieldValue length = new FieldValue(inputObject, "length", TypeUtils.PRIMITIVE_INT_TYPE);
- Invoke reset = new Invoke(arrayWriter, "reset", length);
+ Expression reset = new Invoke(arrayWriter, "reset", length);
if (rawType.getComponentType().isPrimitive()) {
return new ListExpression(
reset, new Invoke(arrayWriter, "fromPrimitiveArray", inputObject), arrayWriter);
@@ -330,30 +356,17 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
value,
arrayWriter,
Objects.requireNonNull(typeRef.getComponentType()),
+ null,
arrayElementField,
new HashSet<>()));
return new ListExpression(reset, forEach, arrayWriter);
}
- } else if (getRawType(typeRef) == Iterable.class) {
- ListFromIterable listFromIterable = new ListFromIterable(inputObject);
- Invoke size = new Invoke(listFromIterable, "size", TypeUtils.PRIMITIVE_INT_TYPE);
- Invoke reset = new Invoke(arrayWriter, "reset", size);
- ForEach forEach =
- new ForEach(
- listFromIterable,
- !TypeUtils.getElementType(typeRef).isPrimitive(),
- (i, value) ->
- serializeFor(
- i,
- value,
- arrayWriter,
- TypeUtils.getElementType(typeRef),
- arrayElementField,
- new HashSet<>()));
- return new ListExpression(reset, forEach, arrayWriter);
- } else { // collection
+ } else {
+ if (getRawType(typeRef) == Iterable.class) {
+ inputObject = new ListFromIterable(inputObject);
+ }
Invoke size = new Invoke(inputObject, "size", TypeUtils.PRIMITIVE_INT_TYPE);
- Invoke reset = new Invoke(arrayWriter, "reset", size);
+ Expression reset = new Invoke(arrayWriter, "reset", size);
ForEach forEach =
new ForEach(
inputObject,
@@ -364,6 +377,7 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
value,
arrayWriter,
TypeUtils.getElementType(typeRef),
+ null,
arrayElementField,
new HashSet<>()));
return new ListExpression(reset, forEach, arrayWriter);
@@ -377,10 +391,10 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
t -> {
String name = ctx.newName("arrayWriter");
ctx.addField(
- ctx.type(BinaryArrayWriter.class),
+ arrayWriterType().getRawType(),
name,
- new NewInstance(arrayWriterTypeToken, arrayDataType, writer));
- return new Reference(name, arrayWriterTypeToken, false);
+ new NewInstance(arrayWriterType(), arrayDataType, writer));
+ return new Reference(name, arrayWriterType(), false);
});
}
@@ -393,6 +407,7 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
Expression writer,
Expression inputObject,
TypeRef<?> typeRef,
+ Field fieldIfKnown,
Expression arrowField) {
StaticInvoke keyArrayField =
new StaticInvoke(
@@ -424,7 +439,8 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
expressions.add(offset, preserve);
Invoke keySet = new Invoke(inputObject, "keySet", keySetType);
- Expression keySerializationExpr = serializeForArray(keySet, writer, keySetType, keyArrayField);
+ Expression keySerializationExpr =
+ serializeForArray(keySet, writer, keySetType, fieldIfKnown, keyArrayField);
expressions.add(keySet, keySerializationExpr);
expressions.add(
@@ -436,7 +452,7 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
Invoke values = new Invoke(inputObject, "values", valuesType);
Expression valueSerializationExpr =
- serializeForArray(values, writer, valuesType, valueArrayField);
+ serializeForArray(values, writer, valuesType, fieldIfKnown, valueArrayField);
expressions.add(values, valueSerializationExpr);
Arithmetic size =
@@ -457,6 +473,7 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
Expression ordinal,
Expression writer,
Expression inputObject,
+ Field fieldIfKnown,
TypeRef<?> typeRef,
Expression structField) {
Class<?> rawType = getRawType(typeRef);
@@ -464,13 +481,11 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
Reference beanEncoder = beanEncoderMap.get(typeRef);
if (beanEncoder == null) {
// janino generics don't add cast, so this `<${type}>` is only for generated code readability
- StaticInvoke schema =
- new StaticInvoke(
- DataTypes.class, "schemaFromStructField", "schema", SCHEMA_TYPE, false, structField);
+ Expression schema = createSchemaFromStructField(structField);
String rowWriterName =
ctx.newName(StringUtils.uncapitalize(rawType.getSimpleName() + "RowWriter"));
- NewInstance newRowWriter = new NewInstance(rowWriterTypeToken, schema, writer);
- ctx.addField(ctx.type(rowWriterTypeToken), rowWriterName, newRowWriter);
+ NewInstance newRowWriter = new NewInstance(rowWriterType(), schema, writer);
+ ctx.addField(ctx.type(rowWriterType()), rowWriterName, newRowWriter);
Preconditions.checkArgument(!codecClassName(rawType).contains("."));
String encoderName = ctx.newName(StringUtils.uncapitalize(codecClassName(rawType)));
@@ -483,30 +498,49 @@ public abstract class BaseBinaryEncoderBuilder extends CodecBuilder {
ExpressionUtils.newObjectArray(schema, newRowWriter, foryRef));
ctx.addField(encoderClass, encoderName, newEncoder);
- rowWriter = new Reference(rowWriterName, rowWriterTypeToken);
+ rowWriter = new Reference(rowWriterName, rowWriterType());
rowWriterMap.put(typeRef, rowWriter);
beanEncoder = new Reference(encoderName, codecTypeRef);
beanEncoderMap.put(typeRef, beanEncoder);
}
rowWriter = rowWriterMap.get(typeRef);
- Invoke reset = new Invoke(rowWriter, "reset");
+ Expression expression =
+ serializeForNotNullBean(ordinal, writer, inputObject, fieldIfKnown, rowWriter, beanEncoder);
+
+ return new If(
+ new Expression.IsNull(inputObject), new Invoke(writer, "setNullAt", ordinal), expression);
+ }
+
+ protected Expression createSchemaFromStructField(Expression structField) {
+ return new StaticInvoke(
+ DataTypes.class, "schemaFromStructField", "schema", SCHEMA_TYPE, false, structField);
+ }
+
+ protected Expression serializeForNotNullBean(
+ Expression ordinal,
+ Expression writer,
+ Expression inputObject,
+ Field fieldIfKnown,
+ Reference rowWriter,
+ Reference beanEncoder) {
+ Invoke reset = beanWriterReset(writer, rowWriter, ordinal);
Invoke offset = new Invoke(writer, "writerIndex", "writerIndex", TypeUtils.PRIMITIVE_INT_TYPE);
Invoke toRow = new Invoke(beanEncoder, "toRow", inputObject);
Arithmetic size =
ExpressionUtils.subtract(
new Invoke(writer, "writerIndex", "writerIndex", TypeUtils.PRIMITIVE_INT_TYPE), offset);
Invoke setOffsetAndSize = new Invoke(writer, "setOffsetAndSize", ordinal, offset, size);
- ListExpression expression =
- new ListExpression(
- offset,
- reset,
- toRow, // reset will change writerIndex. must call reset and toRow in pair.
- size,
- setOffsetAndSize);
+ return new ListExpression(
+ offset,
+ reset, // reset will change writerIndex. must call reset and toRow in pair.
+ toRow,
+ size,
+ setOffsetAndSize);
+ }
- return new If(
- new Expression.IsNull(inputObject), new Invoke(writer, "setNullAt", ordinal), expression);
+ protected Invoke beanWriterReset(Expression writer, Reference rowWriter, Expression ordinal) {
+ return new Invoke(rowWriter, "reset");
}
/**
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/BaseCodecBuilder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BaseCodecBuilder.java
new file mode 100644
index 000000000..581e102a1
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BaseCodecBuilder.java
@@ -0,0 +1,75 @@
+/*
+ * 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.fory.format.encoder;
+
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.fory.Fory;
+import org.apache.fory.format.row.binary.CompactBinaryRow;
+import org.apache.fory.format.row.binary.writer.CompactBinaryRowWriter;
+
+public class BaseCodecBuilder<B extends BaseCodecBuilder<B>> {
+ protected Schema schema;
+ protected int initialBufferSize = 16;
+ protected boolean sizeEmbedded = true;
+ protected Fory fory;
+ protected Encoding codecFormat = DefaultCodecFormat.INSTANCE;
+
+ BaseCodecBuilder(final Schema schema) {
+ this.schema = schema;
+ }
+
+ /** Configure the Fory instance used for embedded binary serialized objects. */
+ public B fory(final Fory fory) {
+ this.fory = fory;
+ return castThis();
+ }
+
+ /** Configure the initial buffer size used when writing a new row. */
+ public B initialBufferSize(final int initialBufferSize) {
+ this.initialBufferSize = initialBufferSize;
+ return castThis();
+ }
+
+ /**
+ * Configure whether the encoder expects the size encoded inline in the data, or provided by
+ * external framing. When true, the default, the encoder will write the size as part of its data,
+ * to be read by the decoder. When false, the data size must be preserved by external framing you
+ * provide.
+ */
+ public B withSizeEmbedded(final boolean sizeEmbedded) {
+ this.sizeEmbedded = sizeEmbedded;
+ return castThis();
+ }
+
+ /**
+ * Configure compact encoding, which is more space efficient than the default encoding, but is not
+ * yet stable. See {@link CompactBinaryRow} for details.
+ */
+ public B compactEncoding() {
+ schema = CompactBinaryRowWriter.sortSchema(schema);
+ codecFormat = CompactCodecFormat.INSTANCE;
+ return castThis();
+ }
+
+ @SuppressWarnings("unchecked")
+ protected B castThis() {
+ return (B) this;
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/BinaryArrayEncoder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BinaryArrayEncoder.java
new file mode 100644
index 000000000..f956310de
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BinaryArrayEncoder.java
@@ -0,0 +1,101 @@
+/*
+ * 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.fory.format.encoder;
+
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.fory.format.row.binary.BinaryArray;
+import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
+import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.memory.MemoryUtils;
+
+class BinaryArrayEncoder<T> implements ArrayEncoder<T> {
+ private final BinaryArrayWriter writer;
+ private final GeneratedArrayEncoder codec;
+ private final boolean sizeEmbedded;
+
+ BinaryArrayEncoder(
+ final BinaryArrayWriter writer,
+ final GeneratedArrayEncoder codec,
+ final boolean sizeEmbedded) {
+ this.writer = writer;
+ this.codec = codec;
+ this.sizeEmbedded = sizeEmbedded;
+ }
+
+ @Override
+ public Field field() {
+ return writer.getField();
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public T fromArray(final BinaryArray array) {
+ return (T) codec.fromArray(array);
+ }
+
+ @Override
+ public BinaryArray toArray(final T obj) {
+ return codec.toArray(obj);
+ }
+
+ @Override
+ public T decode(final MemoryBuffer buffer) {
+ return decode(buffer, sizeEmbedded ? buffer.readInt32() : buffer.remaining());
+ }
+
+ @Override
+ public T decode(final byte[] bytes) {
+ return decode(MemoryUtils.wrap(bytes), bytes.length);
+ }
+
+ T decode(final MemoryBuffer buffer, final int size) {
+ final BinaryArray array = writer.newArray();
+ final int readerIndex = buffer.readerIndex();
+ array.pointTo(buffer, readerIndex, size);
+ buffer.readerIndex(readerIndex + size);
+ return fromArray(array);
+ }
+
+ @Override
+ public byte[] encode(final T obj) {
+ final BinaryArray array = toArray(obj);
+ return writer.getBuffer().getBytes(0, array.getSizeInBytes());
+ }
+
+ @Override
+ public int encode(final MemoryBuffer buffer, final T obj) {
+ final MemoryBuffer prevBuffer = writer.getBuffer();
+ final int writerIndex = buffer.writerIndex();
+ if (sizeEmbedded) {
+ buffer.writeInt32(-1);
+ }
+ try {
+ writer.setBuffer(buffer);
+ toArray(obj);
+ final int size = buffer.writerIndex() - writerIndex;
+ if (sizeEmbedded) {
+ buffer.putInt32(writerIndex, size - 4);
+ }
+ return size;
+ } finally {
+ writer.setBuffer(prevBuffer);
+ }
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/BinaryMapEncoder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BinaryMapEncoder.java
new file mode 100644
index 000000000..b5514d917
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BinaryMapEncoder.java
@@ -0,0 +1,118 @@
+/*
+ * 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.fory.format.encoder;
+
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.fory.format.row.binary.BinaryArray;
+import org.apache.fory.format.row.binary.BinaryMap;
+import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
+import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.memory.MemoryUtils;
+
+class BinaryMapEncoder<M> implements MapEncoder<M> {
+ private final Encoding format;
+ private final Field mapField;
+ private final BinaryArrayWriter valWriter;
+ private final BinaryArrayWriter keyWriter;
+ private final GeneratedMapEncoder codec;
+ private final boolean sizeEmbedded;
+
+ BinaryMapEncoder(
+ final Encoding format,
+ final Field mapField,
+ final BinaryArrayWriter valWriter,
+ final BinaryArrayWriter keyWriter,
+ final GeneratedMapEncoder codec,
+ final boolean sizeEmbedded) {
+ this.format = format;
+ this.mapField = mapField;
+ this.valWriter = valWriter;
+ this.keyWriter = keyWriter;
+ this.codec = codec;
+ this.sizeEmbedded = sizeEmbedded;
+ }
+
+ @Override
+ public Field keyField() {
+ return keyWriter.getField();
+ }
+
+ @Override
+ public Field valueField() {
+ return valWriter.getField();
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public M fromMap(final BinaryArray key, final BinaryArray value) {
+ return (M) codec.fromMap(key, value);
+ }
+
+ @Override
+ public BinaryMap toMap(final M obj) {
+ return codec.toMap(obj);
+ }
+
+ @Override
+ public M decode(final MemoryBuffer buffer) {
+ return decode(buffer, sizeEmbedded ? buffer.readInt32() : buffer.remaining());
+ }
+
+ M decode(final MemoryBuffer buffer, final int size) {
+ final BinaryMap map = format.newMap(mapField);
+ final int readerIndex = buffer.readerIndex();
+ map.pointTo(buffer, readerIndex, size);
+ buffer.readerIndex(readerIndex + size);
+ return fromMap(map);
+ }
+
+ @Override
+ public M decode(final byte[] bytes) {
+ return decode(MemoryUtils.wrap(bytes), bytes.length);
+ }
+
+ @Override
+ public byte[] encode(final M obj) {
+ final BinaryMap map = toMap(obj);
+ return map.getBuf().getBytes(map.getBaseOffset(), map.getSizeInBytes());
+ }
+
+ @Override
+ public int encode(final MemoryBuffer buffer, final M obj) {
+ final MemoryBuffer prevBuffer = keyWriter.getBuffer();
+ final int writerIndex = buffer.writerIndex();
+ if (sizeEmbedded) {
+ buffer.writeInt32(-1);
+ }
+ try {
+ keyWriter.setBuffer(buffer);
+ valWriter.setBuffer(buffer);
+ toMap(obj);
+ final int size = buffer.writerIndex() - writerIndex;
+ if (sizeEmbedded) {
+ buffer.putInt32(writerIndex, size - 4);
+ }
+ return size;
+ } finally {
+ keyWriter.setBuffer(prevBuffer);
+ valWriter.setBuffer(prevBuffer);
+ }
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/BinaryRowEncoder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BinaryRowEncoder.java
new file mode 100644
index 000000000..c9266daa0
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BinaryRowEncoder.java
@@ -0,0 +1,125 @@
+/*
+ * 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.fory.format.encoder;
+
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.fory.exception.ClassNotCompatibleException;
+import org.apache.fory.format.row.binary.BinaryRow;
+import org.apache.fory.format.row.binary.writer.BaseBinaryRowWriter;
+import org.apache.fory.format.type.DataTypes;
+import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.memory.MemoryUtils;
+
+class BinaryRowEncoder<T> implements RowEncoder<T> {
+ private final Schema schema;
+ private final Encoding codecFactory;
+ private final GeneratedRowEncoder codec;
+ private final BaseBinaryRowWriter writer;
+ private final boolean sizeEmbedded;
+ private final long schemaHash;
+ private final MemoryBuffer buffer = MemoryUtils.buffer(16);
+
+ BinaryRowEncoder(
+ final Schema schema,
+ final Encoding codecFactory,
+ final GeneratedRowEncoder codec,
+ final BaseBinaryRowWriter writer,
+ final boolean sizeEmbedded) {
+ this.schema = schema;
+ this.codecFactory = codecFactory;
+ this.codec = codec;
+ this.writer = writer;
+ this.sizeEmbedded = sizeEmbedded;
+ this.schemaHash = DataTypes.computeSchemaHash(schema);
+ }
+
+ @Override
+ public Schema schema() {
+ return schema;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public T fromRow(final BinaryRow row) {
+ return (T) codec.fromRow(row);
+ }
+
+ @Override
+ public BinaryRow toRow(final T obj) {
+ return codec.toRow(obj);
+ }
+
+ @Override
+ public T decode(final MemoryBuffer buffer) {
+ return decode(buffer, sizeEmbedded ? buffer.readInt32() : buffer.remaining());
+ }
+
+ T decode(final MemoryBuffer buffer, final int size) {
+ final long peerSchemaHash = buffer.readInt64();
+ if (peerSchemaHash != schemaHash) {
+ throw new ClassNotCompatibleException(
+ String.format(
+ "Schema is not consistent, encoder schema is %s. "
+ + "self/peer schema hash are %s/%s. "
+ + "Please check writer schema.",
+ schema, schemaHash, peerSchemaHash));
+ }
+ final BinaryRow row = codecFactory.newRow(schema);
+ row.pointTo(buffer, buffer.readerIndex(), size);
+ buffer.increaseReaderIndex(size - 8);
+ return fromRow(row);
+ }
+
+ @Override
+ public T decode(final byte[] bytes) {
+ return decode(MemoryUtils.wrap(bytes), bytes.length);
+ }
+
+ @Override
+ public byte[] encode(final T obj) {
+ buffer.writerIndex(0);
+ buffer.writeInt64(schemaHash);
+ writer.setBuffer(buffer);
+ writer.reset();
+ final BinaryRow row = toRow(obj);
+ return buffer.getBytes(0, 8 + row.getSizeInBytes());
+ }
+
+ @Override
+ public int encode(final MemoryBuffer buffer, final T obj) {
+ final int writerIndex = buffer.writerIndex();
+ if (sizeEmbedded) {
+ buffer.writeInt32(-1);
+ }
+ try {
+ buffer.writeInt64(schemaHash);
+ writer.setBuffer(buffer);
+ writer.reset();
+ codec.toRow(obj);
+ final int size = buffer.writerIndex() - writerIndex;
+ if (sizeEmbedded) {
+ buffer.putInt32(writerIndex, size - 4);
+ }
+ return size;
+ } finally {
+ writer.setBuffer(this.buffer);
+ }
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/BufferResettingArrayEncoder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BufferResettingArrayEncoder.java
new file mode 100644
index 000000000..7202a1e45
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BufferResettingArrayEncoder.java
@@ -0,0 +1,80 @@
+/*
+ * 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.fory.format.encoder;
+
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.fory.format.row.binary.BinaryArray;
+import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
+import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.memory.MemoryUtils;
+
+/**
+ * Wrap an ArrayEncoder to provide a fresh buffer for every toArray operation, when not streaming.
+ */
+class BufferResettingArrayEncoder<T> implements ArrayEncoder<T> {
+
+ private final int initialBufferSize;
+ private final BinaryArrayWriter writer;
+ private final ArrayEncoder<T> encoder;
+
+ BufferResettingArrayEncoder(
+ final int initialBufferSize, final BinaryArrayWriter writer, final ArrayEncoder<T> encoder) {
+ this.initialBufferSize = initialBufferSize;
+ this.writer = writer;
+ this.encoder = encoder;
+ }
+
+ @Override
+ public T decode(final MemoryBuffer buffer) {
+ return encoder.decode(buffer);
+ }
+
+ @Override
+ public T decode(final byte[] bytes) {
+ return encoder.decode(bytes);
+ }
+
+ @Override
+ public byte[] encode(final T obj) {
+ writer.setBuffer(MemoryUtils.buffer(initialBufferSize));
+ return encoder.encode(obj);
+ }
+
+ @Override
+ public int encode(final MemoryBuffer buffer, final T obj) {
+ return encoder.encode(buffer, obj);
+ }
+
+ @Override
+ public Field field() {
+ return encoder.field();
+ }
+
+ @Override
+ public T fromArray(final BinaryArray array) {
+ return encoder.fromArray(array);
+ }
+
+ @Override
+ public BinaryArray toArray(final T obj) {
+ writer.setBuffer(MemoryUtils.buffer(initialBufferSize));
+ return encoder.toArray(obj);
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/BufferResettingMapEncoder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BufferResettingMapEncoder.java
new file mode 100644
index 000000000..d76a13ea4
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BufferResettingMapEncoder.java
@@ -0,0 +1,93 @@
+/*
+ * 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.fory.format.encoder;
+
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.fory.format.row.binary.BinaryArray;
+import org.apache.fory.format.row.binary.BinaryMap;
+import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
+import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.memory.MemoryUtils;
+
+/** Wrap a MapEncoder to provide a fresh buffer for every toMap operation, when not streaming. */
+class BufferResettingMapEncoder<T> implements MapEncoder<T> {
+
+ private final int initialBufferSize;
+ private final BinaryArrayWriter keyWriter;
+ private final BinaryArrayWriter valWriter;
+ private final MapEncoder<T> encoder;
+
+ BufferResettingMapEncoder(
+ final int initialBufferSize,
+ final BinaryArrayWriter keyWriter,
+ final BinaryArrayWriter valWriter,
+ final MapEncoder<T> encoder) {
+ this.initialBufferSize = initialBufferSize;
+ this.keyWriter = keyWriter;
+ this.valWriter = valWriter;
+ this.encoder = encoder;
+ }
+
+ @Override
+ public Field keyField() {
+ return encoder.keyField();
+ }
+
+ @Override
+ public Field valueField() {
+ return encoder.valueField();
+ }
+
+ @Override
+ public T fromMap(final BinaryArray keyArray, final BinaryArray valArray) {
+ return encoder.fromMap(keyArray, valArray);
+ }
+
+ @Override
+ public BinaryMap toMap(final T obj) {
+ final MemoryBuffer buffer = MemoryUtils.buffer(initialBufferSize);
+ keyWriter.setBuffer(buffer);
+ valWriter.setBuffer(buffer);
+ return encoder.toMap(obj);
+ }
+
+ @Override
+ public T decode(final MemoryBuffer buffer) {
+ return encoder.decode(buffer);
+ }
+
+ @Override
+ public T decode(final byte[] bytes) {
+ return encoder.decode(bytes);
+ }
+
+ @Override
+ public byte[] encode(final T obj) {
+ final MemoryBuffer buffer = MemoryUtils.buffer(initialBufferSize);
+ keyWriter.setBuffer(buffer);
+ valWriter.setBuffer(buffer);
+ return encoder.encode(obj);
+ }
+
+ @Override
+ public int encode(final MemoryBuffer buffer, final T obj) {
+ return encoder.encode(buffer, obj);
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/BufferResettingRowEncoder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BufferResettingRowEncoder.java
new file mode 100644
index 000000000..9106e4ca0
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/BufferResettingRowEncoder.java
@@ -0,0 +1,78 @@
+/*
+ * 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.fory.format.encoder;
+
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.fory.format.row.binary.BinaryRow;
+import org.apache.fory.format.row.binary.writer.BaseBinaryRowWriter;
+import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.memory.MemoryUtils;
+
+/** Wrap a RowEncoder to provide a fresh buffer for every toRow operation, when not streaming. */
+class BufferResettingRowEncoder<T> implements RowEncoder<T> {
+
+ private final int initialBufferSize;
+ private final BaseBinaryRowWriter writer;
+ private final RowEncoder<T> encoder;
+
+ BufferResettingRowEncoder(
+ final int initialBufferSize, final BaseBinaryRowWriter writer, final RowEncoder<T> encoder) {
+ this.initialBufferSize = initialBufferSize;
+ this.writer = writer;
+ this.encoder = encoder;
+ }
+
+ @Override
+ public Schema schema() {
+ return encoder.schema();
+ }
+
+ @Override
+ public T fromRow(final BinaryRow row) {
+ return encoder.fromRow(row);
+ }
+
+ @Override
+ public BinaryRow toRow(final T obj) {
+ writer.setBuffer(MemoryUtils.buffer(initialBufferSize));
+ writer.reset();
+ return encoder.toRow(obj);
+ }
+
+ @Override
+ public T decode(final MemoryBuffer buffer) {
+ return encoder.decode(buffer);
+ }
+
+ @Override
+ public T decode(final byte[] bytes) {
+ return encoder.decode(bytes);
+ }
+
+ @Override
+ public byte[] encode(final T obj) {
+ return encoder.encode(obj);
+ }
+
+ @Override
+ public int encode(final MemoryBuffer buffer, final T obj) {
+ return encoder.encode(buffer, obj);
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/CompactArrayEncoderBuilder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/CompactArrayEncoderBuilder.java
new file mode 100644
index 000000000..65f8508e3
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/CompactArrayEncoderBuilder.java
@@ -0,0 +1,61 @@
+/*
+ * 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.fory.format.encoder;
+
+import org.apache.fory.codegen.Expression;
+import org.apache.fory.codegen.Expression.Invoke;
+import org.apache.fory.codegen.Expression.Reference;
+import org.apache.fory.format.row.binary.writer.BaseBinaryRowWriter;
+import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
+import org.apache.fory.format.row.binary.writer.CompactBinaryArrayWriter;
+import org.apache.fory.format.row.binary.writer.CompactBinaryRowWriter;
+import org.apache.fory.reflect.TypeRef;
+
+class CompactArrayEncoderBuilder extends ArrayEncoderBuilder {
+ public CompactArrayEncoderBuilder(final TypeRef<?> clsType, final TypeRef<?> beanType) {
+ super(clsType, beanType);
+ }
+
+ @Override
+ protected Invoke beanWriterReset(
+ final Expression writer, final Reference rowWriter, final Expression ordinal) {
+ return new Invoke(writer, "resetFor", rowWriter, ordinal);
+ }
+
+ @Override
+ protected TypeRef<? extends BaseBinaryRowWriter> rowWriterType() {
+ return TypeRef.of(CompactBinaryRowWriter.class);
+ }
+
+ @Override
+ protected TypeRef<? extends BinaryArrayWriter> arrayWriterType() {
+ return TypeRef.of(CompactBinaryArrayWriter.class);
+ }
+
+ @Override
+ protected String codecSuffix() {
+ return "CompactCodec";
+ }
+
+ @Override
+ protected Expression createSchemaFromStructField(final Expression schema) {
+ return CompactRowEncoderBuilder.sortSchema(super.createSchemaFromStructField(schema));
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/CompactCodecFormat.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/CompactCodecFormat.java
new file mode 100644
index 000000000..359ec8fc4
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/CompactCodecFormat.java
@@ -0,0 +1,93 @@
+/*
+ * 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.fory.format.encoder;
+
+import java.util.Collection;
+import java.util.Map;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.fory.format.row.binary.BinaryArray;
+import org.apache.fory.format.row.binary.BinaryMap;
+import org.apache.fory.format.row.binary.BinaryRow;
+import org.apache.fory.format.row.binary.CompactBinaryArray;
+import org.apache.fory.format.row.binary.CompactBinaryMap;
+import org.apache.fory.format.row.binary.CompactBinaryRow;
+import org.apache.fory.format.row.binary.writer.BaseBinaryRowWriter;
+import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
+import org.apache.fory.format.row.binary.writer.CompactBinaryArrayWriter;
+import org.apache.fory.format.row.binary.writer.CompactBinaryRowWriter;
+import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.reflect.TypeRef;
+
+enum CompactCodecFormat implements Encoding {
+ INSTANCE;
+
+ @Override
+ public BaseBinaryRowWriter newWriter(final Schema schema) {
+ return new CompactBinaryRowWriter(schema);
+ }
+
+ @Override
+ public BaseBinaryRowWriter newWriter(final Schema schema, final MemoryBuffer buffer) {
+ return new CompactBinaryRowWriter(schema, buffer);
+ }
+
+ @Override
+ public BinaryArrayWriter newArrayWriter(final Field field) {
+ return new CompactBinaryArrayWriter(field);
+ }
+
+ @Override
+ public BinaryArrayWriter newArrayWriter(final Field field, final MemoryBuffer buffer) {
+ return new CompactBinaryArrayWriter(field, buffer);
+ }
+
+ @Override
+ public RowEncoderBuilder newRowEncoder(final TypeRef<?> beanType) {
+ return new CompactRowEncoderBuilder(beanType);
+ }
+
+ @Override
+ public ArrayEncoderBuilder newArrayEncoder(
+ final TypeRef<? extends Collection<?>> collectionType, final TypeRef<?> elementType) {
+ return new CompactArrayEncoderBuilder(collectionType, elementType);
+ }
+
+ @Override
+ public MapEncoderBuilder newMapEncoder(
+ final TypeRef<? extends Map<?, ?>> mapType, final TypeRef<?> beanToken) {
+ return new CompactMapEncoderBuilder(mapType, beanToken);
+ }
+
+ @Override
+ public BinaryRow newRow(final Schema schema) {
+ return new CompactBinaryRow(schema);
+ }
+
+ @Override
+ public BinaryArray newArray(final Field field) {
+ return new CompactBinaryArray(field);
+ }
+
+ @Override
+ public BinaryMap newMap(final Field field) {
+ return new CompactBinaryMap(field);
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/CompactMapEncoderBuilder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/CompactMapEncoderBuilder.java
new file mode 100644
index 000000000..be3d206d5
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/CompactMapEncoderBuilder.java
@@ -0,0 +1,69 @@
+/*
+ * 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.fory.format.encoder;
+
+import org.apache.fory.codegen.Expression;
+import org.apache.fory.codegen.Expression.Invoke;
+import org.apache.fory.codegen.Expression.Reference;
+import org.apache.fory.format.row.binary.BinaryMap;
+import org.apache.fory.format.row.binary.CompactBinaryMap;
+import org.apache.fory.format.row.binary.writer.BaseBinaryRowWriter;
+import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
+import org.apache.fory.format.row.binary.writer.CompactBinaryArrayWriter;
+import org.apache.fory.format.row.binary.writer.CompactBinaryRowWriter;
+import org.apache.fory.reflect.TypeRef;
+
+class CompactMapEncoderBuilder extends MapEncoderBuilder {
+
+ public CompactMapEncoderBuilder(final TypeRef<?> clsType, final TypeRef<?> beanType) {
+ super(clsType, beanType);
+ }
+
+ @Override
+ protected Invoke beanWriterReset(
+ final Expression writer, final Reference rowWriter, final Expression ordinal) {
+ return new Invoke(writer, "resetFor", rowWriter, ordinal);
+ }
+
+ @Override
+ protected TypeRef<? extends BaseBinaryRowWriter> rowWriterType() {
+ return TypeRef.of(CompactBinaryRowWriter.class);
+ }
+
+ @Override
+ protected TypeRef<? extends BinaryArrayWriter> arrayWriterType() {
+ return TypeRef.of(CompactBinaryArrayWriter.class);
+ }
+
+ @Override
+ protected TypeRef<? extends BinaryMap> binaryMapType() {
+ return TypeRef.of(CompactBinaryMap.class);
+ }
+
+ @Override
+ protected String codecSuffix() {
+ return "CompactCodec";
+ }
+
+ @Override
+ protected Expression createSchemaFromStructField(final Expression structField) {
+ return CompactRowEncoderBuilder.sortSchema(super.createSchemaFromStructField(structField));
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/CompactRowEncoderBuilder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/CompactRowEncoderBuilder.java
new file mode 100644
index 000000000..5d6e389e6
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/CompactRowEncoderBuilder.java
@@ -0,0 +1,127 @@
+/*
+ * 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.fory.format.encoder;
+
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.fory.codegen.Expression;
+import org.apache.fory.codegen.Expression.Invoke;
+import org.apache.fory.codegen.Expression.ListExpression;
+import org.apache.fory.codegen.Expression.Reference;
+import org.apache.fory.codegen.Expression.StaticInvoke;
+import org.apache.fory.format.row.binary.writer.BaseBinaryRowWriter;
+import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
+import org.apache.fory.format.row.binary.writer.CompactBinaryArrayWriter;
+import org.apache.fory.format.row.binary.writer.CompactBinaryRowWriter;
+import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.reflect.TypeRef;
+
+/** Expression builder for building compact row encoder class. */
+class CompactRowEncoderBuilder extends RowEncoderBuilder {
+ public CompactRowEncoderBuilder(final TypeRef<?> beanType) {
+ super(beanType);
+ }
+
+ @Override
+ protected Schema inferSchema(final TypeRef<?> beanType) {
+ return CompactBinaryRowWriter.sortSchema(super.inferSchema(beanType));
+ }
+
+ @Override
+ protected String codecSuffix() {
+ return "CompactCodec";
+ }
+
+ @Override
+ protected TypeRef<? extends BaseBinaryRowWriter> rowWriterType() {
+ return TypeRef.of(CompactBinaryRowWriter.class);
+ }
+
+ @Override
+ protected TypeRef<? extends BinaryArrayWriter> arrayWriterType() {
+ return TypeRef.of(CompactBinaryArrayWriter.class);
+ }
+
+ @Override
+ protected Expression serializeForNotNullBean(
+ final Expression ordinal,
+ final Expression writer,
+ final Expression inputObject,
+ final Field fieldIfKnown,
+ final Reference rowWriter,
+ final Reference beanEncoder) {
+ if (fieldIfKnown == null || CompactBinaryRowWriter.fixedWidthFor(fieldIfKnown) == -1) {
+ return super.serializeForNotNullBean(
+ ordinal, writer, inputObject, fieldIfKnown, rowWriter, beanEncoder);
+ }
+ // resetFor will change writerIndex. must call reset and toRow in pair.
+ final Invoke reset = beanWriterReset(writer, rowWriter, ordinal);
+ final Invoke toRow = new Invoke(beanEncoder, "toRow", inputObject);
+ return new ListExpression(reset, toRow);
+ }
+
+ @Override
+ protected Invoke beanWriterReset(
+ final Expression writer, final Reference rowWriter, final Expression ordinal) {
+ return new Invoke(writer, "resetFor", rowWriter, ordinal);
+ }
+
+ @Override
+ protected Expression serializeForArrayByWriter(
+ final Expression inputObject,
+ final Expression arrayWriter,
+ final TypeRef<?> typeRef,
+ final Field fieldIfKnown,
+ final Expression arrowField) {
+ final Expression result =
+ super.serializeForArrayByWriter(
+ inputObject, arrayWriter, typeRef, fieldIfKnown, arrowField);
+ if (fieldIfKnown == null
+ || CompactBinaryRowWriter.fixedWidthFor(itemType(fieldIfKnown)) == -1) {
+ return result;
+ }
+ return new ListExpression(
+ result,
+ new Invoke(
+ Invoke.inlineInvoke(arrayWriter, "getBuffer", TypeRef.of(MemoryBuffer.class)),
+ "writerIndex"),
+ arrayWriter);
+ }
+
+ private static Field itemType(final Field fieldIfKnown) {
+ return fieldIfKnown.getChildren().get(0);
+ }
+
+ @Override
+ protected Expression createSchemaFromStructField(final Expression structField) {
+ return sortSchema(super.createSchemaFromStructField(structField));
+ }
+
+ static Expression sortSchema(final Expression schema) {
+ return new StaticInvoke(
+ CompactBinaryRowWriter.class,
+ "sortSchema",
+ "sortedSchema",
+ TypeRef.of(Schema.class),
+ false,
+ true,
+ schema);
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/DefaultCodecFormat.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/DefaultCodecFormat.java
new file mode 100644
index 000000000..3b6bc2e31
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/DefaultCodecFormat.java
@@ -0,0 +1,89 @@
+/*
+ * 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.fory.format.encoder;
+
+import java.util.Collection;
+import java.util.Map;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.fory.format.row.binary.BinaryArray;
+import org.apache.fory.format.row.binary.BinaryMap;
+import org.apache.fory.format.row.binary.BinaryRow;
+import org.apache.fory.format.row.binary.writer.BaseBinaryRowWriter;
+import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
+import org.apache.fory.format.row.binary.writer.BinaryRowWriter;
+import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.reflect.TypeRef;
+
+enum DefaultCodecFormat implements Encoding {
+ INSTANCE;
+
+ @Override
+ public BaseBinaryRowWriter newWriter(final Schema schema) {
+ return new BinaryRowWriter(schema);
+ }
+
+ @Override
+ public BaseBinaryRowWriter newWriter(final Schema schema, final MemoryBuffer buffer) {
+ return new BinaryRowWriter(schema, buffer);
+ }
+
+ @Override
+ public BinaryArrayWriter newArrayWriter(final Field field) {
+ return new BinaryArrayWriter(field);
+ }
+
+ @Override
+ public BinaryArrayWriter newArrayWriter(final Field field, final MemoryBuffer buffer) {
+ return new BinaryArrayWriter(field, buffer);
+ }
+
+ @Override
+ public RowEncoderBuilder newRowEncoder(final TypeRef<?> beanClass) {
+ return new RowEncoderBuilder(beanClass);
+ }
+
+ @Override
+ public ArrayEncoderBuilder newArrayEncoder(
+ final TypeRef<? extends Collection<?>> collectionType, final TypeRef<?> elementType) {
+ return new ArrayEncoderBuilder(collectionType, elementType);
+ }
+
+ @Override
+ public MapEncoderBuilder newMapEncoder(
+ final TypeRef<? extends Map<?, ?>> mapType, final TypeRef<?> beanToken) {
+ return new MapEncoderBuilder(mapType, beanToken);
+ }
+
+ @Override
+ public BinaryRow newRow(final Schema schema) {
+ return new BinaryRow(schema);
+ }
+
+ @Override
+ public BinaryArray newArray(final Field field) {
+ return new BinaryArray(field);
+ }
+
+ @Override
+ public BinaryMap newMap(final Field field) {
+ return new BinaryMap(field);
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoder.java
index 3216b7ca1..f56b7fd48 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoder.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoder.java
@@ -29,12 +29,22 @@ import org.apache.fory.memory.MemoryBuffer;
* @param <T> type of value
*/
public interface Encoder<T> {
-
+ /** Decode a buffer. */
T decode(MemoryBuffer buffer);
+ /** Decode a byte array. */
T decode(byte[] bytes);
+ /** Encode to a byte array. */
byte[] encode(T obj);
- void encode(MemoryBuffer buffer, T obj);
+ /** Encode to a buffer. Returns number of bytes written to the buffer. */
+ int encode(MemoryBuffer buffer, T obj);
+
+ /** Encode to a buffer. Returns a sliced buffer view of bytes written. */
+ default MemoryBuffer encodeSlice(final MemoryBuffer buffer, final T obj) {
+ final int initialIndex = buffer.writerIndex();
+ final int size = encode(buffer, obj);
+ return buffer.slice(initialIndex, size);
+ }
}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoders.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoders.java
index 9cd2accc9..b03ea52e6 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoders.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoders.java
@@ -28,26 +28,17 @@ import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import org.apache.arrow.util.Preconditions;
-import org.apache.arrow.vector.types.pojo.Field;
-import org.apache.arrow.vector.types.pojo.Schema;
import org.apache.fory.Fory;
+import org.apache.fory.builder.CodecBuilder;
import org.apache.fory.codegen.CodeGenerator;
import org.apache.fory.codegen.CompileUnit;
import org.apache.fory.collection.Tuple2;
-import org.apache.fory.exception.ClassNotCompatibleException;
-import org.apache.fory.format.row.binary.BinaryArray;
-import org.apache.fory.format.row.binary.BinaryMap;
-import org.apache.fory.format.row.binary.BinaryRow;
-import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
import org.apache.fory.format.row.binary.writer.BinaryRowWriter;
import org.apache.fory.format.type.CustomTypeEncoderRegistry;
import org.apache.fory.format.type.CustomTypeRegistration;
-import org.apache.fory.format.type.DataTypes;
import org.apache.fory.format.type.TypeInference;
import org.apache.fory.logging.Logger;
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.type.TypeResolutionContext;
import org.apache.fory.type.TypeUtils;
@@ -60,6 +51,22 @@ import org.apache.fory.type.TypeUtils;
public class Encoders {
private static final Logger LOG = LoggerFactory.getLogger(Encoders.class);
+ /** Build a row codec with configurable options through a builder. */
+ public static <T> RowCodecBuilder<T> buildBeanCodec(Class<T> beanClass) {
+ return new RowCodecBuilder<>(beanClass);
+ }
+
+ /** Build an array codec with configurable options through a builder. */
+ public static <C extends Collection<?>> ArrayCodecBuilder<C> buildArrayCodec(
+ TypeRef<C> collectionType) {
+ return new ArrayCodecBuilder<>(collectionType);
+ }
+
+ /** Build a map codec with configurable options through a builder. */
+ public static <M extends Map<?, ?>> MapCodecBuilder<M> buildMapCodec(TypeRef<M> mapType) {
+ return new MapCodecBuilder<>(mapType);
+ }
+
public static <T> RowEncoder<T> bean(Class<T> beanClass) {
return bean(beanClass, 16);
}
@@ -73,48 +80,7 @@ public class Encoders {
}
public static <T> RowEncoder<T> bean(Class<T> beanClass, Fory fory, int initialBufferSize) {
- Schema schema = TypeInference.inferSchema(beanClass);
- BinaryRowWriter writer = new BinaryRowWriter(schema);
- RowEncoder<T> encoder = bean(beanClass, writer, fory);
- return new RowEncoder<T>() {
-
- @Override
- public Schema schema() {
- return encoder.schema();
- }
-
- @Override
- public T fromRow(BinaryRow row) {
- return encoder.fromRow(row);
- }
-
- @Override
- public BinaryRow toRow(T obj) {
- writer.setBuffer(MemoryUtils.buffer(initialBufferSize));
- writer.reset();
- return encoder.toRow(obj);
- }
-
- @Override
- public T decode(MemoryBuffer buffer) {
- return encoder.decode(buffer);
- }
-
- @Override
- public T decode(byte[] bytes) {
- return encoder.decode(bytes);
- }
-
- @Override
- public byte[] encode(T obj) {
- return encoder.encode(obj);
- }
-
- @Override
- public void encode(MemoryBuffer buffer, T obj) {
- encoder.encode(buffer, obj);
- }
- };
+ return buildBeanCodec(beanClass).fory(fory).initialBufferSize(initialBufferSize).build().get();
}
public static <T> RowEncoder<T> bean(Class<T> beanClass, BinaryRowWriter writer) {
@@ -142,92 +108,7 @@ public class Encoders {
* </ul>
*/
public static <T> RowEncoder<T> bean(Class<T> beanClass, BinaryRowWriter writer, Fory fory) {
- Schema schema = writer.getSchema();
-
- try {
- Class<?> rowCodecClass = loadOrGenRowCodecClass(beanClass);
- Object references = new Object[] {schema, writer, fory};
- GeneratedRowEncoder codec =
- rowCodecClass
- .asSubclass(GeneratedRowEncoder.class)
- .getConstructor(Object[].class)
- .newInstance(references);
- long schemaHash = DataTypes.computeSchemaHash(schema);
-
- return new RowEncoder<T>() {
- private final MemoryBuffer buffer = MemoryUtils.buffer(16);
-
- @Override
- public Schema schema() {
- return schema;
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public T fromRow(BinaryRow row) {
- return (T) codec.fromRow(row);
- }
-
- @Override
- public BinaryRow toRow(T obj) {
- return codec.toRow(obj);
- }
-
- @Override
- public T decode(MemoryBuffer buffer) {
- return decode(buffer, buffer.readInt32());
- }
-
- public T decode(MemoryBuffer buffer, int size) {
- long peerSchemaHash = buffer.readInt64();
- if (peerSchemaHash != schemaHash) {
- throw new ClassNotCompatibleException(
- String.format(
- "Schema is not consistent, encoder schema is %s. "
- + "self/peer schema hash are %s/%s. "
- + "Please check writer schema.",
- schema, schemaHash, peerSchemaHash));
- }
- BinaryRow row = new BinaryRow(schema);
- row.pointTo(buffer, buffer.readerIndex(), size);
- buffer.increaseReaderIndex(size - 8);
- return fromRow(row);
- }
-
- @Override
- public T decode(byte[] bytes) {
- return decode(MemoryUtils.wrap(bytes), bytes.length);
- }
-
- @Override
- public byte[] encode(T obj) {
- buffer.writerIndex(0);
- buffer.writeInt64(schemaHash);
- writer.setBuffer(buffer);
- writer.reset();
- BinaryRow row = toRow(obj);
- return buffer.getBytes(0, 8 + row.getSizeInBytes());
- }
-
- @Override
- public void encode(MemoryBuffer buffer, T obj) {
- int writerIndex = buffer.writerIndex();
- buffer.writeInt32(-1);
- try {
- buffer.writeInt64(schemaHash);
- writer.setBuffer(buffer);
- writer.reset();
- toRow(obj);
- buffer.putInt32(writerIndex, buffer.writerIndex() - writerIndex - 4);
- } finally {
- writer.setBuffer(this.buffer);
- }
- }
- };
- } catch (Exception e) {
- String msg = String.format("Create encoder failed, \nbeanClass: %s", beanClass);
- throw new EncoderException(msg, e);
- }
+ return buildBeanCodec(beanClass).fory(fory).buildForWriter().apply(writer);
}
/**
@@ -273,167 +154,13 @@ public class Encoders {
* @param <T> T is a array type, can be a nested list type.
* @return
*/
- public static <T extends Collection> ArrayEncoder<T> arrayEncoder(TypeRef<T> token) {
+ public static <T extends Collection<?>> ArrayEncoder<T> arrayEncoder(TypeRef<T> token) {
return arrayEncoder(token, null);
}
- public static <T extends Collection> ArrayEncoder<T> arrayEncoder(TypeRef<T> token, Fory fory) {
- Schema schema = TypeInference.inferSchema(token, false);
- Field field = DataTypes.fieldOfSchema(schema, 0);
- BinaryArrayWriter writer = new BinaryArrayWriter(field);
-
- Set<TypeRef<?>> set = new HashSet<>();
- findBeanToken(token, set);
- if (set.isEmpty()) {
- throw new IllegalArgumentException("can not find bean class.");
- }
-
- TypeRef<?> typeRef = null;
- for (TypeRef<?> tt : set) {
- typeRef = set.iterator().next();
- Encoders.loadOrGenRowCodecClass(getRawType(tt));
- }
- ArrayEncoder<T> encoder = arrayEncoder(token, typeRef, writer, fory);
- return new ArrayEncoder<T>() {
-
- @Override
- public Field field() {
- return encoder.field();
- }
-
- @Override
- public T fromArray(BinaryArray array) {
- return encoder.fromArray(array);
- }
-
- @Override
- public BinaryArray toArray(T obj) {
- return encoder.toArray(obj);
- }
-
- @Override
- public T decode(MemoryBuffer buffer) {
- return encoder.decode(buffer);
- }
-
- @Override
- public T decode(byte[] bytes) {
- return encoder.decode(bytes);
- }
-
- @Override
- public byte[] encode(T obj) {
- return encoder.encode(obj);
- }
-
- @Override
- public void encode(MemoryBuffer buffer, T obj) {
- encoder.encode(buffer, obj);
- }
- };
- }
-
- /**
- * The underlying implementation uses array, only supported {@link Collection} format, because
- * generic type such as List is erased to simply List, so a bean class input param is required.
- *
- * @return
- */
- public static <T extends Collection, B> ArrayEncoder<T> arrayEncoder(
- Class<? extends Collection> arrayCls, Class<B> elementType) {
- Preconditions.checkNotNull(elementType);
-
- return (ArrayEncoder<T>) arrayEncoder(TypeUtils.collectionOf(elementType), null);
- }
-
- /**
- * Creates an encoder for Java Bean of type T.
- *
- * <p>T must be publicly accessible.
- *
- * <p>supported types for java bean field: - primitive types: boolean, int, double, etc. - boxed
- * types: Boolean, Integer, Double, etc. - String - java.math.BigDecimal, java.math.BigInteger -
- * time related: java.sql.Date, java.sql.Timestamp, java.time.LocalDate, java.time.Instant -
- * collection types: only array and java.util.List currently, map support is in progress - nested
- * java bean.
- */
- public static <T extends Collection, B> ArrayEncoder<T> arrayEncoder(
- TypeRef<? extends Collection> arrayToken,
- TypeRef<B> elementType,
- BinaryArrayWriter writer,
- Fory fory) {
- Field field = writer.getField();
- try {
- Class<?> rowCodecClass = loadOrGenArrayCodecClass(arrayToken, elementType);
- Object references = new Object[] {field, writer, fory};
- GeneratedArrayEncoder codec =
- rowCodecClass
- .asSubclass(GeneratedArrayEncoder.class)
- .getConstructor(Object[].class)
- .newInstance(references);
-
- return new ArrayEncoder<T>() {
-
- @Override
- public Field field() {
- return field;
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public T fromArray(BinaryArray array) {
- return (T) codec.fromArray(array);
- }
-
- @Override
- public BinaryArray toArray(T obj) {
- return codec.toArray(obj);
- }
-
- @Override
- public T decode(MemoryBuffer buffer) {
- return decode(buffer, buffer.readInt32());
- }
-
- public T decode(MemoryBuffer buffer, int size) {
- BinaryArray array = new BinaryArray(field);
- int readerIndex = buffer.readerIndex();
- array.pointTo(buffer, readerIndex, size);
- buffer.readerIndex(readerIndex + size);
- return fromArray(array);
- }
-
- @Override
- public T decode(byte[] bytes) {
- return decode(MemoryUtils.wrap(bytes), bytes.length);
- }
-
- @Override
- public byte[] encode(T obj) {
- BinaryArray array = toArray(obj);
- return writer.getBuffer().getBytes(0, array.getSizeInBytes());
- }
-
- @Override
- public void encode(MemoryBuffer buffer, T obj) {
- MemoryBuffer prevBuffer = writer.getBuffer();
- int writerIndex = buffer.writerIndex();
- buffer.writeInt32(-1);
- try {
- writer.setBuffer(buffer);
- BinaryArray array = toArray(obj);
- int size = buffer.writerIndex() - writerIndex - 4;
- assert size == array.getSizeInBytes();
- buffer.putInt32(writerIndex, size);
- } finally {
- writer.setBuffer(prevBuffer);
- }
- }
- };
- } catch (Exception e) {
- String msg = String.format("Create encoder failed, \nelementType: %s", elementType);
- throw new EncoderException(msg, e);
- }
+ public static <T extends Collection<?>> ArrayEncoder<T> arrayEncoder(
+ TypeRef<T> token, Fory fory) {
+ return buildArrayCodec(token).fory(fory).build().get();
}
/**
@@ -455,6 +182,7 @@ public class Encoders {
*
* @return
*/
+ @SuppressWarnings("unchecked")
public static <T extends Map, K, V> MapEncoder<T> mapEncoder(
Class<? extends Map> mapCls, Class<K> keyType, Class<V> valueType) {
Preconditions.checkNotNull(keyType);
@@ -463,19 +191,21 @@ public class Encoders {
return (MapEncoder<T>) mapEncoder(TypeUtils.mapOf(keyType, valueType), null);
}
- public static <T extends Map> MapEncoder<T> mapEncoder(TypeRef<T> token, Fory fory) {
+ @SuppressWarnings("unchecked")
+ public static <T extends Map<K, V>, K, V> MapEncoder<T> mapEncoder(TypeRef<T> token, Fory fory) {
Preconditions.checkNotNull(token);
- Tuple2<TypeRef<?>, TypeRef<?>> tuple2 = TypeUtils.getMapKeyValueType(token);
+ final Tuple2<TypeRef<?>, TypeRef<?>> tuple2 = TypeUtils.getMapKeyValueType(token);
- Set<TypeRef<?>> set1 = beanSet(tuple2.f0);
- Set<TypeRef<?>> set2 = beanSet(tuple2.f1);
+ final Set<TypeRef<?>> set1 = beanSet(tuple2.f0);
+ final Set<TypeRef<?>> set2 = beanSet(tuple2.f1);
LOG.info("Find beans to load: {}, {}", set1, set2);
- TypeRef<?> keyToken = token4BeanLoad(set1, tuple2.f0);
- TypeRef<?> valToken = token4BeanLoad(set2, tuple2.f1);
+ final TypeRef<K> keyToken =
+ (TypeRef<K>) token4BeanLoad(set1, tuple2.f0, DefaultCodecFormat.INSTANCE);
+ final TypeRef<V> valToken =
+ (TypeRef<V>) token4BeanLoad(set2, tuple2.f1, DefaultCodecFormat.INSTANCE);
- MapEncoder<T> encoder = mapEncoder0(token, keyToken, valToken, fory);
- return createMapEncoder(encoder);
+ return mapEncoder0(token, keyToken, valToken, fory);
}
/**
@@ -489,110 +219,24 @@ public class Encoders {
* collection types: only array and java.util.List currently, map support is in progress - nested
* java bean.
*/
- public static <T extends Map, K, V> MapEncoder<T> mapEncoder(
- TypeRef<? extends Map> mapToken, TypeRef<K> keyToken, TypeRef<V> valToken, Fory fory) {
+ public static <T extends Map<K, V>, K, V> MapEncoder<T> mapEncoder(
+ TypeRef<T> mapToken, TypeRef<K> keyToken, TypeRef<V> valToken, Fory fory) {
Preconditions.checkNotNull(mapToken);
Preconditions.checkNotNull(keyToken);
Preconditions.checkNotNull(valToken);
-
- Set<TypeRef<?>> set1 = beanSet(keyToken);
- Set<TypeRef<?>> set2 = beanSet(valToken);
- LOG.info("Find beans to load: {}, {}", set1, set2);
-
- token4BeanLoad(set1, keyToken);
- token4BeanLoad(set2, valToken);
-
return mapEncoder0(mapToken, keyToken, valToken, fory);
}
- private static <T extends Map, K, V> MapEncoder<T> mapEncoder0(
- TypeRef<? extends Map> mapToken, TypeRef<K> keyToken, TypeRef<V> valToken, Fory fory) {
+ private static <T extends Map<K, V>, K, V> MapEncoder<T> mapEncoder0(
+ TypeRef<T> mapToken, TypeRef<K> keyToken, TypeRef<V> valToken, Fory fory) {
Preconditions.checkNotNull(mapToken);
Preconditions.checkNotNull(keyToken);
Preconditions.checkNotNull(valToken);
+ return buildMapCodec(mapToken).fory(fory).build().get();
+ }
- Schema schema = TypeInference.inferSchema(mapToken, false);
- Field field = DataTypes.fieldOfSchema(schema, 0);
- Field keyField = DataTypes.keyArrayFieldForMap(field);
- Field valField = DataTypes.itemArrayFieldForMap(field);
- BinaryArrayWriter keyWriter = new BinaryArrayWriter(keyField);
- BinaryArrayWriter valWriter = new BinaryArrayWriter(valField, keyWriter.getBuffer());
- try {
- Class<?> rowCodecClass = loadOrGenMapCodecClass(mapToken, keyToken, valToken);
- Object references = new Object[] {keyField, valField, keyWriter, valWriter, fory, field};
- GeneratedMapEncoder codec =
- rowCodecClass
- .asSubclass(GeneratedMapEncoder.class)
- .getConstructor(Object[].class)
- .newInstance(references);
-
- return new MapEncoder<T>() {
- @Override
- public Field keyField() {
- return keyField;
- }
-
- @Override
- public Field valueField() {
- return valField;
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public T fromMap(BinaryArray key, BinaryArray value) {
- return (T) codec.fromMap(key, value);
- }
-
- @Override
- public BinaryMap toMap(T obj) {
- return codec.toMap(obj);
- }
-
- @Override
- public T decode(MemoryBuffer buffer) {
- return decode(buffer, buffer.readInt32());
- }
-
- public T decode(MemoryBuffer buffer, int size) {
- BinaryMap map = new BinaryMap(field);
- int readerIndex = buffer.readerIndex();
- map.pointTo(buffer, readerIndex, size);
- buffer.readerIndex(readerIndex + size);
- return fromMap(map);
- }
-
- @Override
- public T decode(byte[] bytes) {
- return decode(MemoryUtils.wrap(bytes), bytes.length);
- }
-
- @Override
- public byte[] encode(T obj) {
- BinaryMap map = toMap(obj);
- return map.getBuf().getBytes(map.getBaseOffset(), map.getSizeInBytes());
- }
-
- @Override
- public void encode(MemoryBuffer buffer, T obj) {
- MemoryBuffer prevBuffer = keyWriter.getBuffer();
- int writerIndex = buffer.writerIndex();
- buffer.writeInt32(-1);
- try {
- keyWriter.setBuffer(buffer);
- valWriter.setBuffer(buffer);
- toMap(obj);
- buffer.putInt32(writerIndex, buffer.writerIndex() - writerIndex - 4);
- } finally {
- keyWriter.setBuffer(prevBuffer);
- valWriter.setBuffer(prevBuffer);
- }
- }
- };
- } catch (Exception e) {
- String msg =
- String.format("Create encoder failed, \nkeyType: %s, valueType: %s", keyToken, valToken);
- throw new EncoderException(msg, e);
- }
+ static void loadMapCodecs(TypeRef<?> type, Encoding codecFactory) {
+ token4BeanLoad(beanSet(type), type, codecFactory);
}
private static Set<TypeRef<?>> beanSet(TypeRef<?> token) {
@@ -606,62 +250,18 @@ public class Encoders {
return set;
}
- private static TypeRef<?> token4BeanLoad(Set<TypeRef<?>> set, TypeRef<?> init) {
+ private static TypeRef<?> token4BeanLoad(
+ Set<TypeRef<?>> set, TypeRef<?> init, Encoding codecFactory) {
TypeRef<?> keyToken = init;
for (TypeRef<?> tt : set) {
keyToken = tt;
- Encoders.loadOrGenRowCodecClass(getRawType(tt));
+ Encoders.loadOrGenRowCodecClass(getRawType(tt), codecFactory);
LOG.info("bean {} load finished", getRawType(tt));
}
return keyToken;
}
- private static <T> MapEncoder<T> createMapEncoder(MapEncoder<T> encoder) {
- return new MapEncoder<T>() {
-
- @Override
- public Field keyField() {
- return encoder.keyField();
- }
-
- @Override
- public Field valueField() {
- return encoder.valueField();
- }
-
- @Override
- public T fromMap(BinaryArray key, BinaryArray value) {
- return encoder.fromMap(key, value);
- }
-
- @Override
- public BinaryMap toMap(T obj) {
- return encoder.toMap(obj);
- }
-
- @Override
- public T decode(MemoryBuffer buffer) {
- return encoder.decode(buffer);
- }
-
- @Override
- public T decode(byte[] bytes) {
- return encoder.decode(bytes);
- }
-
- @Override
- public byte[] encode(T obj) {
- return encoder.encode(obj);
- }
-
- @Override
- public void encode(MemoryBuffer buffer, T obj) {
- encoder.encode(buffer, obj);
- }
- };
- }
-
- private static void findBeanToken(TypeRef<?> typeRef, java.util.Set<TypeRef<?>> set) {
+ static void findBeanToken(TypeRef<?> typeRef, final Set<TypeRef<?>> set) {
TypeResolutionContext typeCtx =
new TypeResolutionContext(CustomTypeEncoderRegistry.customTypeHandler(), true);
Set<TypeRef<?>> visited = new LinkedHashSet<>();
@@ -696,7 +296,7 @@ public class Encoders {
}
}
- public static Class<?> loadOrGenRowCodecClass(Class<?> beanClass) {
+ static Class<?> loadOrGenRowCodecClass(Class<?> beanClass, Encoding codecFactory) {
Set<Class<?>> classes =
TypeUtils.listBeansRecursiveInclusive(
beanClass,
@@ -704,12 +304,12 @@ public class Encoders {
if (classes.isEmpty()) {
return null;
}
- LOG.info("Create RowCodec for classes {}", classes);
+ LOG.info("Create codec for classes {}", classes);
CompileUnit[] compileUnits =
classes.stream()
.map(
cls -> {
- RowEncoderBuilder codecBuilder = new RowEncoderBuilder(cls);
+ final CodecBuilder codecBuilder = codecFactory.newRowEncoder(TypeRef.of(cls));
// use genCodeFunc to avoid gen code repeatedly
return new CompileUnit(
CodeGenerator.getPackage(cls),
@@ -720,14 +320,14 @@ public class Encoders {
return loadCls(compileUnits);
}
- private static <B> Class<?> loadOrGenArrayCodecClass(
- TypeRef<? extends Collection> arrayCls, TypeRef<B> elementType) {
+ static <B> Class<?> loadOrGenArrayCodecClass(
+ TypeRef<? extends Collection<?>> arrayCls, TypeRef<B> elementType, Encoding codecFactory) {
LOG.info("Create ArrayCodec for classes {}", elementType);
Class<?> cls = getRawType(elementType);
// class name prefix
String prefix = TypeInference.inferTypeName(arrayCls);
- ArrayEncoderBuilder codecBuilder = new ArrayEncoderBuilder(arrayCls, elementType);
+ ArrayEncoderBuilder codecBuilder = codecFactory.newArrayEncoder(arrayCls, elementType);
CompileUnit compileUnit =
new CompileUnit(
CodeGenerator.getPackage(cls),
@@ -737,8 +337,11 @@ public class Encoders {
return loadCls(compileUnit);
}
- private static <K, V> Class<?> loadOrGenMapCodecClass(
- TypeRef<? extends Map> mapCls, TypeRef<K> keyToken, TypeRef<V> valueToken) {
+ static <K, V> Class<?> loadOrGenMapCodecClass(
+ TypeRef<? extends Map<?, ?>> mapCls,
+ TypeRef<K> keyToken,
+ TypeRef<V> valueToken,
+ Encoding codecFactory) {
LOG.info("Create MapCodec for classes {}, {}", keyToken, valueToken);
boolean keyIsBean = TypeUtils.isBean(keyToken);
boolean valIsBean = TypeUtils.isBean(valueToken);
@@ -757,7 +360,7 @@ public class Encoders {
// class name prefix
String prefix = TypeInference.inferTypeName(mapCls);
- MapEncoderBuilder codecBuilder = new MapEncoderBuilder(mapCls, beanToken);
+ MapEncoderBuilder codecBuilder = codecFactory.newMapEncoder(mapCls, beanToken);
CompileUnit compileUnit =
new CompileUnit(
CodeGenerator.getPackage(cls),
@@ -774,7 +377,7 @@ public class Encoders {
String className = compileUnit[0].getQualifiedClassName();
try {
return classLoader.loadClass(className);
- } catch (ClassNotFoundException e) {
+ } catch (final ClassNotFoundException e) {
throw new IllegalStateException("Impossible because we just compiled class", e);
}
}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoding.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoding.java
new file mode 100644
index 000000000..3f73a869f
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoding.java
@@ -0,0 +1,55 @@
+/*
+ * 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.fory.format.encoder;
+
+import java.util.Collection;
+import java.util.Map;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.fory.format.row.binary.BinaryArray;
+import org.apache.fory.format.row.binary.BinaryMap;
+import org.apache.fory.format.row.binary.BinaryRow;
+import org.apache.fory.format.row.binary.writer.BaseBinaryRowWriter;
+import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
+import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.reflect.TypeRef;
+
+interface Encoding {
+ BaseBinaryRowWriter newWriter(Schema schema);
+
+ BaseBinaryRowWriter newWriter(Schema schema, MemoryBuffer buffer);
+
+ BinaryArrayWriter newArrayWriter(Field field);
+
+ BinaryArrayWriter newArrayWriter(Field field, MemoryBuffer buffer);
+
+ RowEncoderBuilder newRowEncoder(TypeRef<?> beanType);
+
+ ArrayEncoderBuilder newArrayEncoder(
+ TypeRef<? extends Collection<?>> collectionType, TypeRef<?> elementType);
+
+ MapEncoderBuilder newMapEncoder(TypeRef<? extends Map<?, ?>> mapType, TypeRef<?> beanToken);
+
+ BinaryRow newRow(Schema schema);
+
+ BinaryArray newArray(Field field);
+
+ BinaryMap newMap(Field field);
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/GeneratedRowEncoder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/GeneratedRowEncoder.java
index 3fd763623..9af0f20f7 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/encoder/GeneratedRowEncoder.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/GeneratedRowEncoder.java
@@ -19,12 +19,13 @@
package org.apache.fory.format.encoder;
+import org.apache.fory.annotation.Internal;
import org.apache.fory.builder.Generated;
import org.apache.fory.format.row.binary.BinaryRow;
/** A row format codec for java bean. */
+@Internal
public interface GeneratedRowEncoder extends Generated {
-
BinaryRow toRow(Object obj);
Object fromRow(BinaryRow row);
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/MapCodecBuilder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/MapCodecBuilder.java
new file mode 100644
index 000000000..6b4bdbbf7
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/MapCodecBuilder.java
@@ -0,0 +1,108 @@
+/*
+ * 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.fory.format.encoder;
+
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
+import java.util.Map;
+import java.util.function.BiFunction;
+import java.util.function.Supplier;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
+import org.apache.fory.format.type.DataTypes;
+import org.apache.fory.format.type.TypeInference;
+import org.apache.fory.reflect.TypeRef;
+import org.apache.fory.type.TypeUtils;
+import org.apache.fory.util.ExceptionUtils;
+
+public class MapCodecBuilder<M extends Map<?, ?>> extends BaseCodecBuilder<MapCodecBuilder<M>> {
+
+ private final TypeRef<M> mapType;
+ private final Field field;
+ private final Field keyField;
+ private final Field valField;
+ private final TypeRef<?> keyType;
+ private final TypeRef<?> valType;
+
+ MapCodecBuilder(final TypeRef<M> mapType) {
+ super(TypeInference.inferSchema(mapType, false));
+ this.mapType = mapType;
+ field = DataTypes.fieldOfSchema(schema, 0);
+ keyField = DataTypes.keyArrayFieldForMap(field);
+ valField = DataTypes.itemArrayFieldForMap(field);
+ final var kvType = TypeUtils.getMapKeyValueType(mapType);
+ keyType = kvType.f0;
+ valType = kvType.f1;
+ }
+
+ public Supplier<MapEncoder<M>> build() {
+ loadMapInnerCodecs();
+ final var mapEncoderFactory = generatedMapEncoder();
+ return new Supplier<MapEncoder<M>>() {
+ @Override
+ public MapEncoder<M> get() {
+ final BinaryArrayWriter keyWriter = codecFormat.newArrayWriter(keyField);
+ final BinaryArrayWriter valWriter =
+ codecFormat.newArrayWriter(valField, keyWriter.getBuffer());
+ final var codec = mapEncoderFactory.apply(keyWriter, valWriter);
+ return new BufferResettingMapEncoder<>(
+ initialBufferSize,
+ keyWriter,
+ valWriter,
+ new BinaryMapEncoder<M>(codecFormat, field, valWriter, keyWriter, codec, sizeEmbedded));
+ }
+ };
+ }
+
+ private void loadMapInnerCodecs() {
+ Encoders.loadMapCodecs(keyType, codecFormat);
+ Encoders.loadMapCodecs(valType, codecFormat);
+ }
+
+ BiFunction<BinaryArrayWriter, BinaryArrayWriter, GeneratedMapEncoder> generatedMapEncoder() {
+ final Class<?> arrayCodecClass =
+ Encoders.loadOrGenMapCodecClass(mapType, keyType, valType, codecFormat);
+
+ final MethodHandle constructorHandle;
+ try {
+ final var constructor =
+ arrayCodecClass.asSubclass(GeneratedMapEncoder.class).getConstructor(Object[].class);
+ constructorHandle =
+ MethodHandles.lookup()
+ .unreflectConstructor(constructor)
+ .asType(MethodType.methodType(GeneratedMapEncoder.class, Object[].class));
+ } catch (final NoSuchMethodException | IllegalAccessException e) {
+ throw new EncoderException("Failed to construct array codec for " + mapType, e);
+ }
+ return new BiFunction<BinaryArrayWriter, BinaryArrayWriter, GeneratedMapEncoder>() {
+ @Override
+ public GeneratedMapEncoder apply(
+ final BinaryArrayWriter keyWriter, final BinaryArrayWriter valWriter) {
+ final Object[] references = {keyField, valField, keyWriter, valWriter, fory, field};
+ try {
+ return (GeneratedMapEncoder) constructorHandle.invokeExact(references);
+ } catch (final Throwable t) {
+ throw ExceptionUtils.throwAnyway(t);
+ }
+ }
+ };
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/MapEncoderBuilder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/MapEncoderBuilder.java
index 94468406f..974a59ad8 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/encoder/MapEncoderBuilder.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/MapEncoderBuilder.java
@@ -32,7 +32,6 @@ import org.apache.fory.codegen.Expression;
import org.apache.fory.codegen.ExpressionUtils;
import org.apache.fory.format.row.binary.BinaryArray;
import org.apache.fory.format.row.binary.BinaryMap;
-import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
import org.apache.fory.format.type.TypeInference;
import org.apache.fory.logging.Logger;
import org.apache.fory.logging.LoggerFactory;
@@ -52,7 +51,6 @@ public class MapEncoderBuilder extends BaseBinaryEncoderBuilder {
private static final String ROOT_KEY_WRITER_NAME = "keyArrayWriter";
private static final String ROOT_VALUE_WRITER_NAME = "valueArrayWriter";
- private static final TypeRef<Field> ARROW_FIELD_TYPE = TypeRef.of(Field.class);
private final TypeRef<?> mapToken;
public MapEncoderBuilder(Class<?> mapCls, Class<?> keyClass) {
@@ -97,7 +95,7 @@ public class MapEncoderBuilder extends BaseBinaryEncoderBuilder {
"keyArrayWriter",
ROOT_KEY_WRITER_NAME,
"arrayWriterType",
- ctx.type(BinaryArrayWriter.class),
+ ctx.type(arrayWriterType()),
"valueField",
VALUE_FIELD_NAME,
"fieldType",
@@ -105,7 +103,7 @@ public class MapEncoderBuilder extends BaseBinaryEncoderBuilder {
"valueArrayWriter",
ROOT_VALUE_WRITER_NAME,
"arrayWriterType",
- ctx.type(BinaryArrayWriter.class),
+ ctx.type(arrayWriterType()),
"fory",
FORY_NAME,
"foryType",
@@ -116,8 +114,8 @@ public class MapEncoderBuilder extends BaseBinaryEncoderBuilder {
ctx.type(Field.class));
ctx.addField(ctx.type(Field.class), KEY_FIELD_NAME);
ctx.addField(ctx.type(Field.class), VALUE_FIELD_NAME);
- ctx.addField(ctx.type(BinaryArrayWriter.class), ROOT_KEY_WRITER_NAME);
- ctx.addField(ctx.type(BinaryArrayWriter.class), ROOT_VALUE_WRITER_NAME);
+ ctx.addField(ctx.type(arrayWriterType()), ROOT_KEY_WRITER_NAME);
+ ctx.addField(ctx.type(arrayWriterType()), ROOT_VALUE_WRITER_NAME);
ctx.addField(ctx.type(Fory.class), FORY_NAME);
ctx.addField(ctx.type(Field.class), FIELD_NAME);
@@ -158,9 +156,9 @@ public class MapEncoderBuilder extends BaseBinaryEncoderBuilder {
new Expression.Cast(inputObject, mapToken, ctx.newName(getRawType(mapToken)), false, false);
Expression.Reference keyArrayWriter =
- new Expression.Reference(ROOT_KEY_WRITER_NAME, arrayWriterTypeToken, false);
+ new Expression.Reference(ROOT_KEY_WRITER_NAME, arrayWriterType(), false);
Expression.Reference valArrayWriter =
- new Expression.Reference(ROOT_VALUE_WRITER_NAME, arrayWriterTypeToken, false);
+ new Expression.Reference(ROOT_VALUE_WRITER_NAME, arrayWriterType(), false);
Expression.Reference fieldExpr = new Expression.Reference(FIELD_NAME, ARROW_FIELD_TYPE, false);
Expression.Reference keyFieldExpr =
@@ -180,7 +178,7 @@ public class MapEncoderBuilder extends BaseBinaryEncoderBuilder {
expressions.add(
new Expression.Invoke(keyArrayWriter, "writeDirectly", Expression.Literal.ofInt(-1)));
Expression keySerializationExpr =
- serializeForArrayByWriter(keySet, keyArrayWriter, keySetType, keyFieldExpr);
+ serializeForArrayByWriter(keySet, keyArrayWriter, keySetType, null, keyFieldExpr);
Expression.Invoke keyArray =
new Expression.Invoke(keyArrayWriter, "toArray", TypeRef.of(BinaryArray.class));
expressions.add(map);
@@ -195,7 +193,7 @@ public class MapEncoderBuilder extends BaseBinaryEncoderBuilder {
Expression.Invoke values = new Expression.Invoke(map, "values", valuesType);
Expression valueSerializationExpr =
- serializeForArrayByWriter(values, valArrayWriter, valuesType, valFieldExpr);
+ serializeForArrayByWriter(values, valArrayWriter, valuesType, null, valFieldExpr);
Expression.Invoke valArray =
new Expression.Invoke(valArrayWriter, "toArray", TypeRef.of(BinaryArray.class));
@@ -203,11 +201,14 @@ public class MapEncoderBuilder extends BaseBinaryEncoderBuilder {
expressions.add(valArray);
expressions.add(
new Expression.Return(
- new Expression.NewInstance(
- TypeRef.of(BinaryMap.class), keyArray, valArray, fieldExpr)));
+ new Expression.NewInstance(binaryMapType(), keyArray, valArray, fieldExpr)));
return expressions;
}
+ protected TypeRef<? extends BinaryMap> binaryMapType() {
+ return TypeRef.of(BinaryMap.class);
+ }
+
/**
* Returns an expression that deserialize <code>row</code> as a java bean of type {@link
* MapEncoderBuilder#mapToken}.
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/RowCodecBuilder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/RowCodecBuilder.java
new file mode 100644
index 000000000..adbd2922d
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/RowCodecBuilder.java
@@ -0,0 +1,98 @@
+/*
+ * 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.fory.format.encoder;
+
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import org.apache.fory.format.row.binary.writer.BaseBinaryRowWriter;
+import org.apache.fory.format.type.TypeInference;
+import org.apache.fory.util.ExceptionUtils;
+
+public class RowCodecBuilder<T> extends BaseCodecBuilder<RowCodecBuilder<T>> {
+
+ private final Class<T> beanClass;
+
+ RowCodecBuilder(final Class<T> beanClass) {
+ super(TypeInference.inferSchema(beanClass));
+ this.beanClass = beanClass;
+ }
+
+ /**
+ * Create a codec factory with the configuration settings from this builder. The resulting factory
+ * should be re-used if possible for creating subsequent encoder instances. Encoders are not
+ * thread-safe, so create a separate encoder for each thread that uses it. For platform threads
+ * consider {@link ThreadLocal#withInitial(Supplier)}, or get a new encoder instance for each
+ * virtual thread.
+ */
+ public Supplier<RowEncoder<T>> build() {
+ final Function<BaseBinaryRowWriter, RowEncoder<T>> rowEncoderFactory = buildForWriter();
+ return new Supplier<RowEncoder<T>>() {
+ @Override
+ public RowEncoder<T> get() {
+ final BaseBinaryRowWriter writer = codecFormat.newWriter(schema);
+ return new BufferResettingRowEncoder<T>(
+ initialBufferSize, writer, rowEncoderFactory.apply(writer));
+ }
+ };
+ }
+
+ Function<BaseBinaryRowWriter, RowEncoder<T>> buildForWriter() {
+ final Function<BaseBinaryRowWriter, GeneratedRowEncoder> rowEncoderFactory =
+ rowEncoderFactory();
+ return new Function<BaseBinaryRowWriter, RowEncoder<T>>() {
+ @Override
+ public RowEncoder<T> apply(final BaseBinaryRowWriter writer) {
+ return new BinaryRowEncoder<T>(
+ schema, codecFormat, rowEncoderFactory.apply(writer), writer, sizeEmbedded);
+ }
+ };
+ }
+
+ Function<BaseBinaryRowWriter, GeneratedRowEncoder> rowEncoderFactory() {
+ final Class<?> rowCodecClass = Encoders.loadOrGenRowCodecClass(beanClass, codecFormat);
+ MethodHandle constructorHandle;
+ try {
+ final var constructor =
+ rowCodecClass.asSubclass(GeneratedRowEncoder.class).getConstructor(Object[].class);
+ constructorHandle =
+ MethodHandles.lookup()
+ .unreflectConstructor(constructor)
+ .asType(MethodType.methodType(GeneratedRowEncoder.class, Object[].class));
+ } catch (final NoSuchMethodException | IllegalAccessException e) {
+ throw new EncoderException("Failed to construct codec for " + beanClass, e);
+ }
+ return new Function<BaseBinaryRowWriter, GeneratedRowEncoder>() {
+ @Override
+ public GeneratedRowEncoder apply(final BaseBinaryRowWriter writer) {
+ try {
+ final Object[] references = {schema, writer, fory};
+ return (GeneratedRowEncoder) constructorHandle.invokeExact(references);
+ } catch (final ReflectiveOperationException e) {
+ throw new EncoderException("Failed to construct codec for " + beanClass, e);
+ } catch (final Throwable e) {
+ throw ExceptionUtils.throwAnyway(e);
+ }
+ }
+ };
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/RowEncoder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/RowEncoder.java
index 43a7487cb..5eb951518 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/encoder/RowEncoder.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/RowEncoder.java
@@ -22,7 +22,10 @@ package org.apache.fory.format.encoder;
import org.apache.arrow.vector.types.pojo.Schema;
import org.apache.fory.format.row.binary.BinaryRow;
-/** Encoder to encode/decode object to/from row. */
+/**
+ * Encoder to encode/decode object to/from row. A RowEncoder instance is reusable but not
+ * thread-safe.
+ */
public interface RowEncoder<T> extends Encoder<T> {
Schema schema();
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/RowEncoderBuilder.java b/java/fory-format/src/main/java/org/apache/fory/format/encoder/RowEncoderBuilder.java
index 6a3f59922..bcb044e9a 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/encoder/RowEncoderBuilder.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/encoder/RowEncoderBuilder.java
@@ -49,7 +49,6 @@ import org.apache.fory.format.row.binary.BinaryArray;
import org.apache.fory.format.row.binary.BinaryMap;
import org.apache.fory.format.row.binary.BinaryRow;
import org.apache.fory.format.row.binary.BinaryUtils;
-import org.apache.fory.format.row.binary.writer.BinaryRowWriter;
import org.apache.fory.format.type.DataTypes;
import org.apache.fory.format.type.TypeInference;
import org.apache.fory.logging.Logger;
@@ -64,7 +63,7 @@ import org.apache.fory.util.record.RecordUtils;
/** Expression builder for building jit row encoder class. */
@SuppressWarnings("UnstableApiUsage")
-public class RowEncoderBuilder extends BaseBinaryEncoderBuilder {
+class RowEncoderBuilder extends BaseBinaryEncoderBuilder {
private static final Logger LOG = LoggerFactory.getLogger(RowEncoderBuilder.class);
static final String SCHEMA_NAME = "schema";
static final String ROOT_ROW_NAME = "row";
@@ -72,7 +71,7 @@ public class RowEncoderBuilder extends BaseBinaryEncoderBuilder {
private final String className;
private final SortedMap<String, Descriptor> descriptorsMap;
- private final Schema schema;
+ protected final Schema schema;
protected static final String BEAN_CLASS_NAME = "beanClass";
protected Reference beanClassRef = new Reference(BEAN_CLASS_NAME, CLASS_TYPE);
private final CodegenContext generatedBeanImpl;
@@ -86,7 +85,7 @@ public class RowEncoderBuilder extends BaseBinaryEncoderBuilder {
super(new CodegenContext(), beanType);
Preconditions.checkArgument(beanClass.isInterface() || TypeUtils.isBean(beanType, typeCtx));
className = codecClassName(beanClass);
- this.schema = TypeInference.inferSchema(getRawType(beanType));
+ this.schema = inferSchema(beanType);
this.descriptorsMap = Descriptor.getDescriptorsMap(beanClass);
ctx.reserveName(ROOT_ROW_WRITER_NAME);
ctx.reserveName(SCHEMA_NAME);
@@ -114,6 +113,10 @@ public class RowEncoderBuilder extends BaseBinaryEncoderBuilder {
}
}
+ protected Schema inferSchema(TypeRef<?> beanType) {
+ return TypeInference.inferSchema(getRawType(beanType));
+ }
+
@Override
protected String codecSuffix() {
return "RowCodec";
@@ -131,6 +134,7 @@ public class RowEncoderBuilder extends BaseBinaryEncoderBuilder {
// don't addImport(beanClass), because user class may name collide.
// janino don't support generics, so GeneratedCodec has no generics
ctx.implementsInterfaces(ctx.type(GeneratedRowEncoder.class));
+ String rowWriterType = ctx.type(rowWriterType());
String constructorCode =
StringUtils.format(
"${schema} = (${schemaType})${references}[0];\n"
@@ -145,13 +149,13 @@ public class RowEncoderBuilder extends BaseBinaryEncoderBuilder {
"rowWriter",
ROOT_ROW_WRITER_NAME,
"rowWriterType",
- ctx.type(BinaryRowWriter.class),
+ rowWriterType,
"fory",
FORY_NAME,
"foryType",
ctx.type(Fory.class));
ctx.addField(ctx.type(Schema.class), SCHEMA_NAME);
- ctx.addField(ctx.type(BinaryRowWriter.class), ROOT_ROW_WRITER_NAME);
+ ctx.addField(rowWriterType, ROOT_ROW_WRITER_NAME);
ctx.addField(ctx.type(Fory.class), FORY_NAME);
Expression encodeExpr = buildEncodeExpression();
@@ -186,7 +190,7 @@ public class RowEncoderBuilder extends BaseBinaryEncoderBuilder {
public Expression buildEncodeExpression() {
Reference inputObject = new Reference(ROOT_OBJECT_NAME, TypeUtils.OBJECT_TYPE, false);
Expression bean = tryCastIfPublic(inputObject, beanType);
- Reference writer = new Reference(ROOT_ROW_WRITER_NAME, rowWriterTypeToken, false);
+ Reference writer = new Reference(ROOT_ROW_WRITER_NAME, rowWriterType(), false);
Reference schemaExpr = new Reference(SCHEMA_NAME, schemaTypeToken, false);
CustomCodec<?, ?> customCodec = customTypeHandler.findCodec(beanClass, beanClass);
@@ -198,16 +202,17 @@ public class RowEncoderBuilder extends BaseBinaryEncoderBuilder {
Expression.ListExpression expressions = new Expression.ListExpression();
// schema field's name must correspond to descriptor's name.
for (int i = 0; i < numFields; i++) {
- Descriptor d = getDescriptorByFieldName(schema.getFields().get(i).getName());
+ Field field = schema.getFields().get(i);
+ Descriptor d = getDescriptorByFieldName(field.getName());
Preconditions.checkNotNull(d);
TypeRef<?> fieldType = d.getTypeRef();
Expression fieldValue = getFieldValue(bean, d);
Literal ordinal = Literal.ofInt(i);
- Expression.StaticInvoke field =
+ Expression.StaticInvoke arrowField =
new Expression.StaticInvoke(
DataTypes.class, "fieldOfSchema", ARROW_FIELD_TYPE, false, schemaExpr, ordinal);
Expression fieldExpr =
- serializeFor(ordinal, fieldValue, writer, fieldType, field, new HashSet<>());
+ serializeFor(ordinal, fieldValue, writer, fieldType, field, arrowField, new HashSet<>());
expressions.add(fieldExpr);
}
expressions.add(
@@ -242,7 +247,8 @@ public class RowEncoderBuilder extends BaseBinaryEncoderBuilder {
fieldNames.add(d.getName());
descriptors[i] = d;
TypeRef<?> fieldType = d.getTypeRef();
- Expression.Variable value = new Expression.Variable(d.getName(), nullValue(fieldType));
+ Expression.Variable value =
+ new Expression.Variable("value_" + d.getName(), nullValue(fieldType));
values[i] = value;
expressions.add(value);
Expression.Invoke isNullAt =
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/Getters.java b/java/fory-format/src/main/java/org/apache/fory/format/row/Getters.java
index 6be6c4b80..65d86edd7 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/row/Getters.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/Getters.java
@@ -64,6 +64,6 @@ public interface Getters {
MapData getMap(int ordinal);
default Object get(int ordinal, Field field) {
- return field.getType().accept(new ValueVisitor(this)).apply(ordinal);
+ return field.getType().accept(new ToStringValueVisitor(this)).apply(ordinal);
}
}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/ToStringValueVisitor.java b/java/fory-format/src/main/java/org/apache/fory/format/row/ToStringValueVisitor.java
new file mode 100644
index 000000000..f1cbffff6
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/ToStringValueVisitor.java
@@ -0,0 +1,57 @@
+/*
+ * 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.fory.format.row;
+
+import java.util.function.Function;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+
+class ToStringValueVisitor extends ValueVisitor {
+ ToStringValueVisitor(final Getters getters) {
+ super(getters);
+ }
+
+ @Override
+ public Function<Integer, Object> visit(final ArrowType.Binary type) {
+ return super.visit(type).andThen(this::binaryToString);
+ }
+
+ @Override
+ public Function<Integer, Object> visit(final ArrowType.FixedSizeBinary type) {
+ return super.visit(type).andThen(this::binaryToString);
+ }
+
+ private String binaryToString(final Object obj) {
+ final byte[] bytes = (byte[]) obj;
+ final int clampedLen = Math.min(bytes.length, 256);
+ final StringBuilder result = new StringBuilder(clampedLen * 2 + 5);
+ result.append("0x");
+ for (int i = 0; i < clampedLen; i++) {
+ final String hexStr = Integer.toHexString(bytes[i] & 0xff);
+ if (hexStr.length() == 1) {
+ result.append('0');
+ }
+ result.append(hexStr);
+ }
+ if (bytes.length > clampedLen) {
+ result.append("...");
+ }
+ return result.toString();
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/ValueVisitor.java b/java/fory-format/src/main/java/org/apache/fory/format/row/ValueVisitor.java
index 61001cfa8..487f28100 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/row/ValueVisitor.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/ValueVisitor.java
@@ -88,6 +88,11 @@ class ValueVisitor extends DefaultTypeVisitor<Function<Integer, Object>> {
return getters::getBinary;
}
+ @Override
+ public Function<Integer, Object> visit(ArrowType.FixedSizeBinary type) {
+ return getters::getBinary;
+ }
+
@Override
public Function<Integer, Object> visit(ArrowType.Decimal type) {
return getters::getDecimal;
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/BinaryArray.java b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/BinaryArray.java
index 076f2a1b4..f08bf77e0 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/BinaryArray.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/BinaryArray.java
@@ -47,34 +47,56 @@ import org.apache.fory.util.Preconditions;
*/
public class BinaryArray extends UnsafeTrait implements ArrayData {
private final Field field;
- private final int elementSize;
+ protected final int elementSize;
private MemoryBuffer buffer;
private int numElements;
+ private int bitmapOffset;
private int elementOffset;
private int baseOffset;
private int sizeInBytes;
public BinaryArray(Field field) {
+ this(field, elementSize(field));
+ }
+
+ protected BinaryArray(Field field, int elementSize) {
this.field = field;
+ this.elementSize = elementSize;
+ initializeExtData(1); // Only require at most one slot to cache the schema for array type.
+ }
+
+ private static int elementSize(Field field) {
int width = DataTypes.getTypeWidth(field.getChildren().get(0).getType());
// variable-length element type
if (width < 0) {
- this.elementSize = 8;
+ return 8;
} else {
- this.elementSize = width;
+ return width;
}
- initializeExtData(1); // Only require at most one slot to cache the schema for array type.
}
public void pointTo(MemoryBuffer buffer, int offset, int sizeInBytes) {
+ this.buffer = buffer;
+ this.baseOffset = offset;
// Read the numElements of key array from the aligned first 8 bytes as int.
- final int numElements = (int) buffer.getInt64(offset);
+ final int numElements = readNumElements();
assert numElements >= 0 : "numElements (" + numElements + ") should >= 0";
this.numElements = numElements;
- this.buffer = buffer;
- this.baseOffset = offset;
this.sizeInBytes = sizeInBytes;
- this.elementOffset = offset + calculateHeaderInBytes(this.numElements);
+ this.bitmapOffset = bitmapOffset();
+ this.elementOffset = elementOffset();
+ }
+
+ protected int readNumElements() {
+ return (int) buffer.getInt64(baseOffset);
+ }
+
+ protected int elementOffset() {
+ return baseOffset + calculateHeaderInBytes(this.numElements);
+ }
+
+ protected int bitmapOffset() {
+ return baseOffset + 8;
}
public Field getField() {
@@ -114,19 +136,19 @@ public class BinaryArray extends UnsafeTrait implements ArrayData {
@Override
public void setNotNullAt(int ordinal) {
assertIndexIsValid(ordinal);
- BitUtils.unset(buffer, baseOffset + 8, ordinal);
+ BitUtils.unset(buffer, bitmapOffset, ordinal);
}
@Override
public void setNullAt(int ordinal) {
- BitUtils.set(buffer, baseOffset + 8, ordinal);
+ BitUtils.set(buffer, bitmapOffset, ordinal);
// we assume the corresponding column was already 0
// or will be set to 0 later by the caller side
}
@Override
public boolean isNullAt(int ordinal) {
- return BitUtils.isSet(buffer, baseOffset + 8, ordinal);
+ return BitUtils.isSet(buffer, bitmapOffset, ordinal);
}
@Override
@@ -154,6 +176,10 @@ public class BinaryArray extends UnsafeTrait implements ArrayData {
throw new UnsupportedOperationException();
}
+ public byte[] toBytes() {
+ return buffer.getBytes(baseOffset, sizeInBytes);
+ }
+
public boolean[] toBooleanArray() {
boolean[] values = new boolean[numElements];
buffer.copyToUnsafe(elementOffset, values, Platform.BOOLEAN_ARRAY_OFFSET, numElements);
@@ -200,7 +226,7 @@ public class BinaryArray extends UnsafeTrait implements ArrayData {
public ArrayData copy() {
MemoryBuffer copyBuf = MemoryUtils.buffer(sizeInBytes);
buffer.copyTo(baseOffset, copyBuf, 0, sizeInBytes);
- BinaryArray arrayCopy = new BinaryArray(field);
+ BinaryArray arrayCopy = newArray(field);
arrayCopy.pointTo(copyBuf, 0, sizeInBytes);
return arrayCopy;
}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/BinaryMap.java b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/BinaryMap.java
index 4ff14f412..e23f9abf9 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/BinaryMap.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/BinaryMap.java
@@ -45,8 +45,16 @@ public class BinaryMap implements MapData {
public BinaryMap(Field field) {
this.field = field;
- this.keys = new BinaryArray(DataTypes.keyArrayFieldForMap(field));
- this.values = new BinaryArray(DataTypes.itemArrayFieldForMap(field));
+ this.keys = newArray(DataTypes.keyArrayFieldForMap(field));
+ this.values = newArray(DataTypes.itemArrayFieldForMap(field));
+ }
+
+ protected BinaryArray newArray(Field field) {
+ return new BinaryArray(field);
+ }
+
+ protected BinaryMap newMap(Field field) {
+ return new BinaryMap(field);
}
public BinaryMap(BinaryArray keys, BinaryArray values, Field field) {
@@ -111,7 +119,7 @@ public class BinaryMap implements MapData {
public MapData copy() {
MemoryBuffer copyBuf = MemoryUtils.buffer(sizeInBytes);
buf.copyTo(baseOffset, copyBuf, 0, sizeInBytes);
- BinaryMap mapCopy = new BinaryMap(field);
+ BinaryMap mapCopy = newMap(field);
mapCopy.pointTo(copyBuf, 0, sizeInBytes);
return mapCopy;
}
@@ -129,6 +137,10 @@ public class BinaryMap implements MapData {
buffer.position(pos + sizeInBytes);
}
+ public byte[] toBytes() {
+ return buf.getBytes(baseOffset, sizeInBytes);
+ }
+
@Override
public String toString() {
return "BinaryMap{"
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/BinaryRow.java b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/BinaryRow.java
index ad9f1384f..d2d4d0cba 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/BinaryRow.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/BinaryRow.java
@@ -61,21 +61,25 @@ import org.apache.fory.util.Preconditions;
* </ul>
*/
public class BinaryRow extends UnsafeTrait implements Row {
- private final Schema schema;
- private final int numFields;
- private final int bitmapWidthInBytes;
- private MemoryBuffer buffer;
- private int baseOffset;
- private int sizeInBytes;
+ final Schema schema;
+ final int numFields;
+ final int bitmapWidthInBytes;
+ MemoryBuffer buffer;
+ int baseOffset;
+ int sizeInBytes;
public BinaryRow(Schema schema) {
this.schema = schema;
this.numFields = schema.getFields().size();
Preconditions.checkArgument(numFields > 0);
- this.bitmapWidthInBytes = BitUtils.calculateBitmapWidthInBytes(numFields);
+ this.bitmapWidthInBytes = computeBitmapWidthInBytes();
initializeExtData(numFields);
}
+ protected int computeBitmapWidthInBytes() {
+ return BitUtils.calculateBitmapWidthInBytes(numFields);
+ }
+
public void pointTo(MemoryBuffer buffer, int offset, int sizeInBytes) {
this.buffer = buffer;
this.baseOffset = offset;
@@ -117,20 +121,24 @@ public class BinaryRow extends UnsafeTrait implements Row {
checkArgument(index < numFields, "index (%d) should < %d", index, numFields);
}
+ protected int nullBitmapOffset() {
+ return baseOffset;
+ }
+
@Override
public boolean isNullAt(int ordinal) {
- return BitUtils.isSet(buffer, baseOffset, ordinal);
+ return BitUtils.isSet(buffer, nullBitmapOffset(), ordinal);
}
@Override
public boolean anyNull() {
- return BitUtils.anySet(buffer, baseOffset, bitmapWidthInBytes);
+ return BitUtils.anySet(buffer, nullBitmapOffset(), bitmapWidthInBytes);
}
@Override
public void setNullAt(int ordinal) {
assertIndexIsValid(ordinal);
- BitUtils.set(buffer, baseOffset, ordinal);
+ BitUtils.set(buffer, nullBitmapOffset(), ordinal);
assert DataTypes.getTypeWidth(schema.getFields().get(ordinal).getType()) > 0
: "field[ "
+ ordinal
@@ -144,9 +152,10 @@ public class BinaryRow extends UnsafeTrait implements Row {
buffer.putInt64(getOffset(ordinal), 0);
}
+ @Override
public void setNotNullAt(int ordinal) {
assertIndexIsValid(ordinal);
- BitUtils.unset(buffer, baseOffset, ordinal);
+ BitUtils.unset(buffer, nullBitmapOffset(), ordinal);
}
@Override
@@ -173,11 +182,15 @@ public class BinaryRow extends UnsafeTrait implements Row {
public Row copy() {
MemoryBuffer copyBuf = MemoryUtils.buffer(sizeInBytes);
buffer.copyTo(baseOffset, copyBuf, 0, sizeInBytes);
- BinaryRow copyRow = new BinaryRow(schema);
+ BinaryRow copyRow = rowForCopy();
copyRow.pointTo(copyBuf, 0, sizeInBytes);
return copyRow;
}
+ protected BinaryRow rowForCopy() {
+ return new BinaryRow(schema);
+ }
+
@Override
public String toString() {
if (buffer == null) {
@@ -211,7 +224,7 @@ public class BinaryRow extends UnsafeTrait implements Row {
if (i != 0) {
build.append(',');
}
- build.append(Long.toHexString(buffer.getInt64(baseOffset + i)));
+ build.append(Long.toHexString(buffer.getInt64(nullBitmapOffset() + i)));
}
return build.toString();
}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/CompactBinaryArray.java b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/CompactBinaryArray.java
new file mode 100644
index 000000000..5dc8f258e
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/CompactBinaryArray.java
@@ -0,0 +1,138 @@
+/*
+ * 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.fory.format.row.binary;
+
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.fory.format.row.binary.writer.BinaryWriter;
+import org.apache.fory.format.row.binary.writer.CompactBinaryArrayWriter;
+import org.apache.fory.format.row.binary.writer.CompactBinaryRowWriter;
+import org.apache.fory.format.type.DataTypes;
+import org.apache.fory.memory.MemoryBuffer;
+
+public class CompactBinaryArray extends BinaryArray {
+ private final Field elementField;
+ private final int fixedWidth;
+ private final boolean elementNullable;
+ private int headerInBytes;
+
+ public CompactBinaryArray(final Field field) {
+ super(field, CompactBinaryArrayWriter.elementWidth(field));
+ elementField = field.getChildren().get(0);
+ fixedWidth = CompactBinaryRowWriter.fixedWidthFor(elementField);
+ elementNullable = elementField.isNullable();
+ }
+
+ @Override
+ protected int elementOffset() {
+ return getBaseOffset() + headerInBytes;
+ }
+
+ public static int calculateHeaderInBytes(
+ final int fixedWidth, final int numElements, final boolean elementNullable) {
+ int headerInBytes = 4 + (elementNullable ? (numElements + 7) / 8 : 0);
+ if (fixedWidth % 8 == 0) {
+ headerInBytes = BinaryWriter.roundNumberOfBytesToNearestWord(headerInBytes);
+ }
+ return headerInBytes;
+ }
+
+ @Override
+ protected int bitmapOffset() {
+ return getBaseOffset() + 4;
+ }
+
+ @Override
+ protected int readNumElements() {
+ final var numElements = getBuffer().getInt32(getBaseOffset());
+ headerInBytes = calculateHeaderInBytes(fixedWidth, numElements, elementNullable);
+ return numElements;
+ }
+
+ @Override
+ public boolean isNullAt(final int ordinal) {
+ if (!elementNullable) {
+ return false;
+ }
+ return super.isNullAt(ordinal);
+ }
+
+ @Override
+ public MemoryBuffer getBuffer(final int ordinal) {
+ if (fixedWidth == -1) {
+ return super.getBuffer(ordinal);
+ }
+ if (isNullAt(ordinal)) {
+ return null;
+ }
+ return getBuffer().slice(getOffset(ordinal), elementSize);
+ }
+
+ @Override
+ public byte[] getBinary(final int ordinal) {
+ if (fixedWidth == -1) {
+ return super.getBinary(ordinal);
+ }
+ if (isNullAt(ordinal)) {
+ return null;
+ }
+ final byte[] bytes = new byte[elementSize];
+ getBuffer().get(getOffset(ordinal), bytes, 0, elementSize);
+ return bytes;
+ }
+
+ @Override
+ protected BinaryRow getStruct(final int ordinal, final Field field, final int extDataSlot) {
+ if (isNullAt(ordinal)) {
+ return null;
+ }
+ assert field == elementField;
+ if (fixedWidth == -1) {
+ return super.getStruct(ordinal, field, extDataSlot);
+ }
+ if (extData[extDataSlot] == null) {
+ extData[extDataSlot] = DataTypes.createSchema(field);
+ }
+ final BinaryRow row = newRow((Schema) extData[extDataSlot]);
+ row.pointTo(getBuffer(), getOffset(ordinal), fixedWidth);
+ return row;
+ }
+
+ @Override
+ protected Schema newSchema(final Field field) {
+ return CompactBinaryRowWriter.sortSchema(super.newSchema(field));
+ }
+
+ @Override
+ protected BinaryRow newRow(final Schema schema) {
+ // TODO: don't re-compute fixed offsets
+ return new CompactBinaryRow(schema, CompactBinaryRowWriter.fixedOffsets(schema));
+ }
+
+ @Override
+ protected BinaryArray newArray(final Field field) {
+ return new CompactBinaryArray(field);
+ }
+
+ @Override
+ protected BinaryMap newMap(final Field field) {
+ return new CompactBinaryMap(field);
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoder.java b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/CompactBinaryMap.java
similarity index 59%
copy from java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoder.java
copy to java/fory-format/src/main/java/org/apache/fory/format/row/binary/CompactBinaryMap.java
index 3216b7ca1..49446d3f2 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoder.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/CompactBinaryMap.java
@@ -17,24 +17,26 @@
* under the License.
*/
-package org.apache.fory.format.encoder;
+package org.apache.fory.format.row.binary;
-import org.apache.fory.memory.MemoryBuffer;
+import org.apache.arrow.vector.types.pojo.Field;
-/**
- * The encoding interface for encode/decode object to/from binary. The implementation class must
- * have a constructor with signature {@code Object[] references}, so we can pass any params to
- * codec.
- *
- * @param <T> type of value
- */
-public interface Encoder<T> {
-
- T decode(MemoryBuffer buffer);
+public class CompactBinaryMap extends BinaryMap {
+ public CompactBinaryMap(final Field field) {
+ super(field);
+ }
- T decode(byte[] bytes);
+ public CompactBinaryMap(final BinaryArray keys, final BinaryArray values, final Field field) {
+ super(keys, values, field);
+ }
- byte[] encode(T obj);
+ @Override
+ protected BinaryArray newArray(final Field field) {
+ return new CompactBinaryArray(field);
+ }
- void encode(MemoryBuffer buffer, T obj);
+ @Override
+ protected BinaryMap newMap(final Field field) {
+ return new CompactBinaryMap(field);
+ }
}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/CompactBinaryRow.java b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/CompactBinaryRow.java
new file mode 100644
index 000000000..8cf28e023
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/CompactBinaryRow.java
@@ -0,0 +1,157 @@
+/*
+ * 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.fory.format.row.binary;
+
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.fory.format.row.binary.writer.CompactBinaryRowWriter;
+import org.apache.fory.format.type.DataTypes;
+import org.apache.fory.memory.MemoryBuffer;
+
+/**
+ * A compact version of {@link BinaryRow}. The compact encoding includes additional optimizations:
+ *
+ * <ul>
+ * <li>fixed size binary objects are stored in the fixed size section with no pointer needed
+ * <li>small values can take up fewer than 8 bytes
+ * <li>null bitmap is skipped if all fields are primitive / not-nullable
+ * <li>the header is packed better, with the null-bitmap allowed to borrow alignment padding at
+ * end of fixed section
+ * <li>data alignment is relaxed, which could lead to less performance in very intensive memory
+ * operations
+ * </ul>
+ *
+ * <b>The compact format is still under development and may not be stable yet.</b>
+ */
+public class CompactBinaryRow extends BinaryRow {
+ private final boolean allFieldsNotNullable;
+ private final int[] fixedOffsets;
+ private final int bitmapOffset;
+
+ public CompactBinaryRow(final Schema schema) {
+ this(schema, CompactBinaryRowWriter.fixedOffsets(schema));
+ }
+
+ public CompactBinaryRow(final Schema schema, final int[] fixedOffsets) {
+ super(schema);
+ this.fixedOffsets = fixedOffsets;
+ bitmapOffset = fixedOffsets[fixedOffsets.length - 1];
+ allFieldsNotNullable = CompactBinaryRowWriter.allNotNullable(schema.getFields());
+ }
+
+ @Override
+ protected int computeBitmapWidthInBytes() {
+ // cannot use field due to initialization order
+ if (CompactBinaryRowWriter.allNotNullable(schema.getFields())) {
+ return 0;
+ }
+ return CompactBinaryRowWriter.headerBytes(schema);
+ }
+
+ @Override
+ public boolean isNullAt(final int ordinal) {
+ if (allFieldsNotNullable) {
+ return false;
+ }
+ return super.isNullAt(ordinal);
+ }
+
+ // TODO: this should use StableValue once it's available
+ @Override
+ public int getOffset(final int ordinal) {
+ return baseOffset + fixedOffsets[ordinal];
+ }
+
+ @Override
+ public MemoryBuffer getBuffer(final int ordinal) {
+ final int fixedWidthBinary = CompactBinaryRowWriter.fixedWidthFor(schema, ordinal);
+ if (fixedWidthBinary >= 0) {
+ if (isNullAt(ordinal)) {
+ return null;
+ }
+ return getBuffer().slice(getOffset(ordinal), fixedWidthBinary);
+ } else {
+ return super.getBuffer(ordinal);
+ }
+ }
+
+ @Override
+ public byte[] getBinary(final int ordinal) {
+ final int fixedWidthBinary = CompactBinaryRowWriter.fixedWidthFor(schema, ordinal);
+ if (fixedWidthBinary >= 0) {
+ if (isNullAt(ordinal)) {
+ return null;
+ }
+ final byte[] bytes = new byte[fixedWidthBinary];
+ getBuffer().get(getOffset(ordinal), bytes, 0, fixedWidthBinary);
+ return bytes;
+ } else {
+ return super.getBinary(ordinal);
+ }
+ }
+
+ @Override
+ protected BinaryRow getStruct(final int ordinal, final Field field, final int extDataSlot) {
+ if (isNullAt(ordinal)) {
+ return null;
+ }
+ final int fixedWidthBinary = CompactBinaryRowWriter.fixedWidthFor(schema, ordinal);
+ if (fixedWidthBinary == -1) {
+ return super.getStruct(ordinal, field, extDataSlot);
+ }
+ if (extData[extDataSlot] == null) {
+ extData[extDataSlot] = DataTypes.createSchema(field);
+ }
+ final BinaryRow row = newRow((Schema) extData[extDataSlot]);
+ row.pointTo(getBuffer().slice(getOffset(ordinal), fixedWidthBinary), 0, fixedWidthBinary);
+ return row;
+ }
+
+ @Override
+ protected Schema newSchema(final Field field) {
+ return CompactBinaryRowWriter.sortSchema(super.newSchema(field));
+ }
+
+ @Override
+ protected BinaryRow newRow(final Schema schema) {
+ // TODO: avoid re-computing these offsets
+ return new CompactBinaryRow(schema, CompactBinaryRowWriter.fixedOffsets(schema));
+ }
+
+ @Override
+ protected BinaryArray newArray(final Field field) {
+ return new CompactBinaryArray(field);
+ }
+
+ @Override
+ protected BinaryMap newMap(final Field field) {
+ return new CompactBinaryMap(field);
+ }
+
+ @Override
+ protected int nullBitmapOffset() {
+ return baseOffset + bitmapOffset;
+ }
+
+ @Override
+ protected BinaryRow rowForCopy() {
+ return new CompactBinaryRow(schema, fixedOffsets);
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/UnsafeTrait.java b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/UnsafeTrait.java
index ad7447e8b..a2b2393b7 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/UnsafeTrait.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/UnsafeTrait.java
@@ -36,7 +36,7 @@ import org.apache.fory.memory.MemoryBuffer;
/** Internal to binary row format to reuse code, don't use it in anywhere else. */
abstract class UnsafeTrait implements Getters, Setters {
- private Object[] extData;
+ protected Object[] extData;
abstract MemoryBuffer getBuffer();
@@ -65,46 +65,55 @@ abstract class UnsafeTrait implements Getters, Setters {
// ####################### getters #######################
// ###########################################################
+ @Override
public boolean getBoolean(int ordinal) {
assertIndexIsValid(ordinal);
return getBuffer().getBoolean(getOffset(ordinal));
}
+ @Override
public byte getByte(int ordinal) {
assertIndexIsValid(ordinal);
return getBuffer().getByte(getOffset(ordinal));
}
+ @Override
public short getInt16(int ordinal) {
assertIndexIsValid(ordinal);
return getBuffer().getInt16(getOffset(ordinal));
}
+ @Override
public int getInt32(int ordinal) {
assertIndexIsValid(ordinal);
return getBuffer().getInt32(getOffset(ordinal));
}
+ @Override
public long getInt64(int ordinal) {
assertIndexIsValid(ordinal);
return getBuffer().getInt64(getOffset(ordinal));
}
+ @Override
public float getFloat32(int ordinal) {
assertIndexIsValid(ordinal);
return getBuffer().getFloat32(getOffset(ordinal));
}
+ @Override
public double getFloat64(int ordinal) {
assertIndexIsValid(ordinal);
return getBuffer().getFloat64(getOffset(ordinal));
}
+ @Override
public int getDate(int ordinal) {
assertIndexIsValid(ordinal);
return getBuffer().getInt32(getOffset(ordinal));
}
+ @Override
public long getTimestamp(int ordinal) {
assertIndexIsValid(ordinal);
return getBuffer().getInt64(getOffset(ordinal));
@@ -157,7 +166,7 @@ abstract class UnsafeTrait implements Getters, Setters {
* @param extDataSlot the ext data slot used to cache the schema for the struct.
* @return the binary row representation of the struct.
*/
- BinaryRow getStruct(int ordinal, Field field, int extDataSlot) {
+ protected BinaryRow getStruct(int ordinal, Field field, int extDataSlot) {
if (isNullAt(ordinal)) {
return null;
}
@@ -165,13 +174,21 @@ abstract class UnsafeTrait implements Getters, Setters {
final int relativeOffset = (int) (offsetAndSize >> 32);
final int size = (int) offsetAndSize;
if (extData[extDataSlot] == null) {
- extData[extDataSlot] = DataTypes.createSchema(field);
+ extData[extDataSlot] = newSchema(field);
}
- BinaryRow row = new BinaryRow((Schema) extData[extDataSlot]);
+ BinaryRow row = newRow((Schema) extData[extDataSlot]);
row.pointTo(getBuffer(), getBaseOffset() + relativeOffset, size);
return row;
}
+ protected Schema newSchema(Field field) {
+ return DataTypes.createSchema(field);
+ }
+
+ protected BinaryRow newRow(Schema schema) {
+ return new BinaryRow(schema);
+ }
+
BinaryArray getArray(int ordinal, Field field) {
if (isNullAt(ordinal)) {
return null;
@@ -179,11 +196,15 @@ abstract class UnsafeTrait implements Getters, Setters {
final long offsetAndSize = getInt64(ordinal);
final int relativeOffset = (int) (offsetAndSize >> 32);
final int size = (int) offsetAndSize;
- BinaryArray array = new BinaryArray(field);
+ BinaryArray array = newArray(field);
array.pointTo(getBuffer(), getBaseOffset() + relativeOffset, size);
return array;
}
+ protected BinaryArray newArray(Field field) {
+ return new BinaryArray(field);
+ }
+
BinaryMap getMap(int ordinal, Field field) {
if (isNullAt(ordinal)) {
return null;
@@ -191,11 +212,15 @@ abstract class UnsafeTrait implements Getters, Setters {
final long offsetAndSize = getInt64(ordinal);
final int relativeOffset = (int) (offsetAndSize >> 32);
final int size = (int) offsetAndSize;
- BinaryMap map = new BinaryMap(field);
+ BinaryMap map = newMap(field);
map.pointTo(getBuffer(), getBaseOffset() + relativeOffset, size);
return map;
}
+ protected BinaryMap newMap(Field field) {
+ return new BinaryMap(field);
+ }
+
// ###########################################################
// ####################### setters #######################
// ###########################################################
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryRowWriter.java b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BaseBinaryRowWriter.java
similarity index 61%
copy from java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryRowWriter.java
copy to java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BaseBinaryRowWriter.java
index c46538881..6e5ede8bd 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryRowWriter.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BaseBinaryRowWriter.java
@@ -19,8 +19,6 @@
package org.apache.fory.format.row.binary.writer;
-import static org.apache.fory.memory.BitUtils.calculateBitmapWidthInBytes;
-
import java.math.BigDecimal;
import org.apache.arrow.vector.types.pojo.ArrowType;
import org.apache.arrow.vector.types.pojo.Schema;
@@ -36,31 +34,46 @@ import org.apache.fory.memory.MemoryUtils;
* <p>Must call {@code reset(buffer)}/{@code reset(buffer, offset)} before use this writer to write
* a new row.
*/
-public class BinaryRowWriter extends BinaryWriter {
+public abstract class BaseBinaryRowWriter extends BinaryWriter {
private final Schema schema;
- private final int headerInBytes;
- // fixed width size: bitmap + fixed width region
- // variable-length region follows startOffset + fixedSize
- private final int fixedSize;
- public BinaryRowWriter(Schema schema) {
+ public BaseBinaryRowWriter(Schema schema) {
super(MemoryUtils.buffer(schema.getFields().size() * 32), 0);
super.startIndex = 0;
this.schema = schema;
- this.headerInBytes = calculateBitmapWidthInBytes(schema.getFields().size());
- this.fixedSize = headerInBytes + schema.getFields().size() * 8;
}
- public BinaryRowWriter(Schema schema, BinaryWriter writer) {
+ protected BaseBinaryRowWriter(Schema schema, int bitMapOffset) {
+ super(MemoryUtils.buffer(schema.getFields().size() * 32), bitMapOffset);
+ super.startIndex = 0;
+ this.schema = schema;
+ }
+
+ public BaseBinaryRowWriter(Schema schema, BinaryWriter writer) {
super(writer.getBuffer(), 0);
writer.children.add(this);
// Since we must call reset before use this writer,
// there's no need to set `super.startIndex = writer.writerIndex();`
this.schema = schema;
- this.headerInBytes = calculateBitmapWidthInBytes(schema.getFields().size());
- this.fixedSize = headerInBytes + schema.getFields().size() * 8;
}
+ protected BaseBinaryRowWriter(Schema schema, MemoryBuffer buffer, int bitMapOffset) {
+ super(buffer, bitMapOffset);
+ this.schema = schema;
+ }
+
+ protected BaseBinaryRowWriter(Schema schema, BinaryWriter writer, int bitMapOffset) {
+ super(writer.getBuffer(), bitMapOffset);
+ writer.children.add(this);
+ // Since we must call reset before use this writer,
+ // there's no need to set `super.startIndex = writer.writerIndex();`
+ this.schema = schema;
+ }
+
+ protected abstract int fixedSize();
+
+ protected abstract int headerInBytes();
+
public Schema getSchema() {
return schema;
}
@@ -75,52 +88,18 @@ public class BinaryRowWriter extends BinaryWriter {
*/
public void reset() {
super.startIndex = buffer.writerIndex();
+ int fixedSize = fixedSize();
grow(fixedSize);
buffer._increaseWriterIndexUnsafe(fixedSize);
- int end = startIndex + headerInBytes;
- for (int i = startIndex; i < end; i += 8) {
- buffer.putInt64(i, 0L);
- }
+ resetHeader();
}
- @Override
- public int getOffset(int ordinal) {
- return startIndex + headerInBytes + (ordinal << 3); // ordinal * 8 = (ordinal << 3)
- }
-
- @Override
- public void write(int ordinal, byte value) {
- final int offset = getOffset(ordinal);
- buffer.putInt64(offset, 0L);
- buffer.putByte(offset, value);
- }
-
- @Override
- public void write(int ordinal, boolean value) {
- final int offset = getOffset(ordinal);
- buffer.putInt64(offset, 0L);
- buffer.putBoolean(offset, value);
- }
-
- @Override
- public void write(int ordinal, short value) {
- final int offset = getOffset(ordinal);
- buffer.putInt64(offset, 0L);
- buffer.putInt16(offset, value);
- }
-
- @Override
- public void write(int ordinal, int value) {
- final int offset = getOffset(ordinal);
- buffer.putInt64(offset, 0L);
- buffer.putInt32(offset, value);
- }
-
- @Override
- public void write(int ordinal, float value) {
- final int offset = getOffset(ordinal);
- buffer.putInt64(offset, 0L);
- buffer.putFloat32(offset, value);
+ protected void resetHeader() {
+ int bitmapStart = startIndex + bytesBeforeBitMap;
+ int end = bitmapStart + headerInBytes();
+ for (int i = bitmapStart; i < end; i += 1) {
+ buffer.putByte(i, 0);
+ }
}
@Override
@@ -131,22 +110,28 @@ public class BinaryRowWriter extends BinaryWriter {
@Override
public void setNullAt(int ordinal) {
super.setNullAt(ordinal);
+ clearValue(ordinal);
+ }
+
+ protected void clearValue(int ordinal) {
write(ordinal, 0L);
}
public BinaryRow getRow() {
- BinaryRow row = new BinaryRow(schema);
+ BinaryRow row = newRow();
int size = size();
row.pointTo(buffer, startIndex, size);
return row;
}
public BinaryRow copyToRow() {
- BinaryRow row = new BinaryRow(schema);
+ BinaryRow row = newRow();
int size = size();
MemoryBuffer buffer = MemoryUtils.buffer(size);
this.buffer.copyTo(startIndex, buffer, 0, size);
row.pointTo(buffer, startIndex, size);
return row;
}
+
+ protected abstract BinaryRow newRow();
}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryArrayWriter.java b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryArrayWriter.java
index 862712172..2c1bb7f43 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryArrayWriter.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryArrayWriter.java
@@ -48,10 +48,10 @@ import org.apache.fory.memory.Platform;
public class BinaryArrayWriter extends BinaryWriter {
public static int MAX_ROUNDED_ARRAY_LENGTH = Integer.MAX_VALUE - 15;
- private final Field field;
- private final int elementSize;
- private int numElements;
- private int headerInBytes;
+ protected final Field field;
+ protected final int elementSize;
+ protected int numElements;
+ protected int headerInBytes;
/** Must call reset before using writer constructed by this constructor. */
public BinaryArrayWriter(Field field) {
@@ -73,17 +73,26 @@ public class BinaryArrayWriter extends BinaryWriter {
}
public BinaryArrayWriter(Field field, MemoryBuffer buffer) {
- super(buffer, 8);
- this.field = field;
+ this(field, buffer, 8, elementWidth(field));
+ }
+
+ private static int elementWidth(Field field) {
int width = DataTypes.getTypeWidth(field.getChildren().get(0).getType());
// variable-length element type
if (width < 0) {
- this.elementSize = 8;
+ return 8;
} else {
- this.elementSize = width;
+ return width;
}
}
+ protected BinaryArrayWriter(
+ Field field, MemoryBuffer buffer, int bytesBeforeBitMap, int elementSize) {
+ super(buffer, bytesBeforeBitMap);
+ this.field = field;
+ this.elementSize = elementSize;
+ }
+
/**
* reset BinaryArrayWriter(ArrayType type, BinaryWriter writer) increase writerIndex, which
* increase writer's writerIndex, we need to record writer's writerIndex before call reset, so we
@@ -91,10 +100,9 @@ public class BinaryArrayWriter extends BinaryWriter {
* will change writerIndex, please use it very carefully</em>.
*/
public void reset(int numElements) {
- super.startIndex = writerIndex();
+ startIndex = writerIndex();
this.numElements = numElements;
- // numElements use 8 byte, nullBitsSizeInBytes use multiple of 8 byte
- this.headerInBytes = BinaryArray.calculateHeaderInBytes(numElements);
+ this.headerInBytes = calculateHeaderInBytes();
long dataSize = numElements * (long) elementSize;
if (dataSize > MAX_ROUNDED_ARRAY_LENGTH) {
throw new UnsupportedOperationException("Can't alloc binary array, it's too big");
@@ -103,23 +111,32 @@ public class BinaryArrayWriter extends BinaryWriter {
buffer.grow(headerInBytes + fixedPartInBytes);
// Write numElements and clear out null bits to header
- // store numElements in header in aligned 8 byte, though numElements is 4 byte int
- buffer.putInt64(startIndex, numElements);
+ int numElementsSize = writeNumElements();
int end = startIndex + headerInBytes;
- for (int i = startIndex + 8; i < end; i += 8) {
- buffer.putInt64(i, 0L);
+ for (int i = startIndex + numElementsSize; i < end; i += 1) {
+ buffer.putByte(i, 0);
}
// fill 0 into reminder part of 8-bytes alignment
for (int i = elementSize * numElements; i < fixedPartInBytes; i++) {
buffer.putByte(startIndex + headerInBytes + i, (byte) 0);
}
+ resetAdvanceWriter(fixedPartInBytes);
+ }
+
+ protected void resetAdvanceWriter(int fixedPartInBytes) {
buffer._increaseWriterIndexUnsafe(headerInBytes + fixedPartInBytes);
}
- private void assertIndexIsValid(int index) {
- assert index >= 0 : "index (" + index + ") should >= 0";
- assert index < numElements : "index (" + index + ") should < " + numElements;
+ protected int writeNumElements() {
+ // store numElements in header in aligned 8 byte, though numElements is 4 byte int
+ buffer.putInt64(startIndex, numElements);
+ return 8;
+ }
+
+ protected int calculateHeaderInBytes() {
+ // numElements use 8 byte, nullBitsSizeInBytes use multiple of 8 byte
+ return BinaryArray.calculateHeaderInBytes(numElements);
}
@Override
@@ -162,6 +179,10 @@ public class BinaryArrayWriter extends BinaryWriter {
writeDecimal(ordinal, value, (ArrowType.Decimal) field.getChildren().get(0).getType());
}
+ protected void primitiveArrayAdvance(int size) {
+ // no need to increasewriterIndex, because reset has already increased writerIndex
+ }
+
private void fromPrimitiveArray(Object arr, int offset, int numElements, Field type) {
if (DataTypes.getTypeId(type.getChildren().get(0).getType())
!= DataTypes.getTypeId(this.field.getChildren().get(0).getType())) {
@@ -171,9 +192,9 @@ public class BinaryArrayWriter extends BinaryWriter {
type.getChildren().get(0).getType(), this.field.getChildren().get(0).getType());
throw new IllegalArgumentException(msg);
}
- buffer.copyFromUnsafe(
- startIndex + headerInBytes, arr, offset, numElements * (long) elementSize);
- // no need to increasewriterIndex, because reset has already increased writerIndex
+ int size = numElements * elementSize;
+ buffer.copyFromUnsafe(startIndex + headerInBytes, arr, offset, size);
+ primitiveArrayAdvance(size);
}
public void fromPrimitiveArray(byte[] arr) {
@@ -206,12 +227,16 @@ public class BinaryArrayWriter extends BinaryWriter {
}
public BinaryArray toArray() {
- BinaryArray array = new BinaryArray(field);
+ BinaryArray array = newArray();
int size = size();
array.pointTo(buffer, startIndex, size);
return array;
}
+ public BinaryArray newArray() {
+ return new BinaryArray(field);
+ }
+
public Field getField() {
return field;
}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryRowWriter.java b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryRowWriter.java
index c46538881..4e48b167a 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryRowWriter.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryRowWriter.java
@@ -21,12 +21,9 @@ package org.apache.fory.format.row.binary.writer;
import static org.apache.fory.memory.BitUtils.calculateBitmapWidthInBytes;
-import java.math.BigDecimal;
-import org.apache.arrow.vector.types.pojo.ArrowType;
import org.apache.arrow.vector.types.pojo.Schema;
import org.apache.fory.format.row.binary.BinaryRow;
import org.apache.fory.memory.MemoryBuffer;
-import org.apache.fory.memory.MemoryUtils;
/**
* Writer to write data into buffer using row format, see {@link BinaryRow}.
@@ -36,51 +33,41 @@ import org.apache.fory.memory.MemoryUtils;
* <p>Must call {@code reset(buffer)}/{@code reset(buffer, offset)} before use this writer to write
* a new row.
*/
-public class BinaryRowWriter extends BinaryWriter {
- private final Schema schema;
+public class BinaryRowWriter extends BaseBinaryRowWriter {
private final int headerInBytes;
// fixed width size: bitmap + fixed width region
// variable-length region follows startOffset + fixedSize
private final int fixedSize;
public BinaryRowWriter(Schema schema) {
- super(MemoryUtils.buffer(schema.getFields().size() * 32), 0);
- super.startIndex = 0;
- this.schema = schema;
- this.headerInBytes = calculateBitmapWidthInBytes(schema.getFields().size());
- this.fixedSize = headerInBytes + schema.getFields().size() * 8;
+ super(schema);
+ headerInBytes = headerBytes(schema);
+ fixedSize = fixedAreaSize();
+ }
+
+ public BinaryRowWriter(Schema schema, MemoryBuffer buffer) {
+ super(schema, buffer, 0);
+ headerInBytes = headerBytes(schema);
+ fixedSize = fixedAreaSize();
}
public BinaryRowWriter(Schema schema, BinaryWriter writer) {
- super(writer.getBuffer(), 0);
- writer.children.add(this);
- // Since we must call reset before use this writer,
- // there's no need to set `super.startIndex = writer.writerIndex();`
- this.schema = schema;
- this.headerInBytes = calculateBitmapWidthInBytes(schema.getFields().size());
- this.fixedSize = headerInBytes + schema.getFields().size() * 8;
+ super(schema, writer);
+ this.headerInBytes = headerBytes(schema);
+ this.fixedSize = fixedAreaSize();
+ }
+
+ private int fixedAreaSize() {
+ return headerInBytes + getSchema().getFields().size() * 8;
}
- public Schema getSchema() {
- return schema;
+ static int headerBytes(Schema schema) {
+ return calculateBitmapWidthInBytes(schema.getFields().size());
}
- /**
- * Call {@code reset()} before write nested row to buffer
- *
- * <p>reset BinaryRowWriter(schema, writer) increase writerIndex, which increase writer's
- * writerIndex, so we need to record writer's writerIndex before call reset, so we can call
- * writer's {@code setOffsetAndSize(int ordinal, int absoluteOffset, int size)}. <em>Reset will
- * change writerIndex, please use it very carefully</em>
- */
- public void reset() {
- super.startIndex = buffer.writerIndex();
- grow(fixedSize);
- buffer._increaseWriterIndexUnsafe(fixedSize);
- int end = startIndex + headerInBytes;
- for (int i = startIndex; i < end; i += 8) {
- buffer.putInt64(i, 0L);
- }
+ @Override
+ protected int headerInBytes() {
+ return headerInBytes;
}
@Override
@@ -124,8 +111,8 @@ public class BinaryRowWriter extends BinaryWriter {
}
@Override
- public void write(int ordinal, BigDecimal value) {
- writeDecimal(ordinal, value, (ArrowType.Decimal) schema.getFields().get(ordinal).getType());
+ protected int fixedSize() {
+ return fixedSize;
}
@Override
@@ -134,19 +121,8 @@ public class BinaryRowWriter extends BinaryWriter {
write(ordinal, 0L);
}
- public BinaryRow getRow() {
- BinaryRow row = new BinaryRow(schema);
- int size = size();
- row.pointTo(buffer, startIndex, size);
- return row;
- }
-
- public BinaryRow copyToRow() {
- BinaryRow row = new BinaryRow(schema);
- int size = size();
- MemoryBuffer buffer = MemoryUtils.buffer(size);
- this.buffer.copyTo(startIndex, buffer, 0, size);
- row.pointTo(buffer, startIndex, size);
- return row;
+ @Override
+ protected BinaryRow newRow() {
+ return new BinaryRow(getSchema());
}
}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryWriter.java b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryWriter.java
index 8e08b054c..c3ecf6ab3 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryWriter.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/BinaryWriter.java
@@ -53,7 +53,7 @@ public abstract class BinaryWriter {
// avoid polymorphic setNullAt/setNotNullAt to inline for performance.
// array use 8 byte for numElements
- private final int bytesBeforeBitMap;
+ protected final int bytesBeforeBitMap;
protected final List<BinaryWriter> children;
protected BinaryWriter(MemoryBuffer buffer, int bytesBeforeBitMap) {
@@ -103,11 +103,11 @@ public abstract class BinaryWriter {
buffer.grow(neededSize);
}
- public final void setOffsetAndSize(int ordinal, int size) {
+ public void setOffsetAndSize(int ordinal, int size) {
setOffsetAndSize(ordinal, buffer.writerIndex(), size);
}
- public final void setOffsetAndSize(int ordinal, int absoluteOffset, int size) {
+ public void setOffsetAndSize(int ordinal, int absoluteOffset, int size) {
final long relativeOffset = absoluteOffset - startIndex;
final long offsetAndSize = (relativeOffset << 32) | (long) size;
write(ordinal, offsetAndSize);
@@ -128,11 +128,11 @@ public abstract class BinaryWriter {
BitUtils.set(buffer, startIndex + bytesBeforeBitMap, ordinal);
}
- public final void setNotNullAt(int ordinal) {
+ public void setNotNullAt(int ordinal) {
BitUtils.unset(buffer, startIndex + bytesBeforeBitMap, ordinal);
}
- public final boolean isNullAt(int ordinal) {
+ public boolean isNullAt(int ordinal) {
return BitUtils.isSet(buffer, startIndex + bytesBeforeBitMap, ordinal);
}
@@ -184,31 +184,36 @@ public abstract class BinaryWriter {
}
/** This operation will increase writerIndex by aligned 8-byte. */
- public final void writeUnaligned(int ordinal, byte[] input, int offset, int numBytes) {
+ public void writeUnaligned(int ordinal, byte[] input, int offset, int numBytes) {
final int roundedSize = roundNumberOfBytesToNearestWord(numBytes);
buffer.grow(roundedSize);
zeroOutPaddingBytes(numBytes);
- buffer.put(buffer.writerIndex(), input, offset, numBytes);
+ buffer.put(bufferWriteIndexFor(ordinal), input, offset, numBytes);
setOffsetAndSize(ordinal, numBytes);
- buffer._increaseWriterIndexUnsafe(roundedSize);
+ if (copyShouldIncreaseWriterIndex(ordinal)) {
+ buffer._increaseWriterIndexUnsafe(roundedSize);
+ }
}
/** This operation will increase writerIndex by aligned 8-byte. */
- public final void writeUnaligned(int ordinal, MemoryBuffer input, int offset, int numBytes) {
+ public void writeUnaligned(int ordinal, MemoryBuffer input, int offset, int numBytes) {
final int roundedSize = roundNumberOfBytesToNearestWord(numBytes);
buffer.grow(roundedSize);
zeroOutPaddingBytes(numBytes);
- buffer.copyFrom(buffer.writerIndex(), input, offset, numBytes);
+ buffer.copyFrom(bufferWriteIndexFor(ordinal), input, offset, numBytes);
setOffsetAndSize(ordinal, numBytes);
- buffer._increaseWriterIndexUnsafe(roundedSize);
+ if (copyShouldIncreaseWriterIndex(ordinal)) {
+ buffer._increaseWriterIndexUnsafe(roundedSize);
+ }
}
- public final void writeAlignedBytes(
- int ordinal, MemoryBuffer input, int baseOffset, int numBytes) {
+ public void writeAlignedBytes(int ordinal, MemoryBuffer input, int baseOffset, int numBytes) {
buffer.grow(numBytes);
- buffer.copyFrom(buffer.writerIndex(), input, baseOffset, numBytes);
+ buffer.copyFrom(bufferWriteIndexFor(ordinal), input, baseOffset, numBytes);
setOffsetAndSize(ordinal, numBytes);
- buffer.increaseWriterIndex(numBytes);
+ if (copyShouldIncreaseWriterIndex(ordinal)) {
+ buffer.increaseWriterIndex(numBytes);
+ }
}
protected final void writeDecimal(int ordinal, BigDecimal value, ArrowType.Decimal type) {
@@ -219,15 +224,28 @@ public abstract class BinaryWriter {
DecimalUtility.writeBigDecimalToArrowBuf(
value, arrowBuf, 0, DecimalUtils.DECIMAL_BYTE_LENGTH);
buffer.copyFromUnsafe(
- writerIndex(), null, arrowBuf.memoryAddress(), DecimalUtils.DECIMAL_BYTE_LENGTH);
+ bufferWriteIndexFor(ordinal),
+ null,
+ arrowBuf.memoryAddress(),
+ DecimalUtils.DECIMAL_BYTE_LENGTH);
arrowBuf.getReferenceManager().release();
setOffsetAndSize(ordinal, writerIndex(), DecimalUtils.DECIMAL_BYTE_LENGTH);
- increaseWriterIndex(DecimalUtils.DECIMAL_BYTE_LENGTH);
+ if (copyShouldIncreaseWriterIndex(ordinal)) {
+ increaseWriterIndex(DecimalUtils.DECIMAL_BYTE_LENGTH);
+ }
} else {
setNullAt(ordinal);
}
}
+ protected int bufferWriteIndexFor(int ordinal) {
+ return buffer.writerIndex();
+ }
+
+ protected boolean copyShouldIncreaseWriterIndex(int ordinal) {
+ return true;
+ }
+
/** write long value to position pointed by current writerIndex. */
public final void writeDirectly(long value) {
buffer.grow(8);
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/CompactBinaryArrayWriter.java b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/CompactBinaryArrayWriter.java
new file mode 100644
index 000000000..0f6b827ca
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/CompactBinaryArrayWriter.java
@@ -0,0 +1,147 @@
+/*
+ * 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.fory.format.row.binary.writer;
+
+import static org.apache.fory.format.row.binary.writer.CompactBinaryRowWriter.fixedWidthFor;
+
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.fory.format.row.binary.BinaryArray;
+import org.apache.fory.format.row.binary.CompactBinaryArray;
+import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.memory.MemoryUtils;
+
+public class CompactBinaryArrayWriter extends BinaryArrayWriter {
+
+ private final int fixedWidth;
+ private final boolean elementNullable;
+
+ /** Must call reset before using writer constructed by this constructor. */
+ public CompactBinaryArrayWriter(final Field field) {
+ // buffer size can grow
+ this(field, MemoryUtils.buffer(64));
+ super.startIndex = 0;
+ }
+
+ /**
+ * Write data to writer's buffer.
+ *
+ * <p>Must call reset before using writer constructed by this constructor
+ */
+ public CompactBinaryArrayWriter(final Field field, final BinaryWriter writer) {
+ this(field, writer.buffer);
+ writer.children.add(this);
+ // Since we must call reset before use this writer,
+ // there's no need to set `super.startIndex = writer.writerIndex();`
+ }
+
+ public CompactBinaryArrayWriter(final Field field, final MemoryBuffer buffer) {
+ super(CompactBinaryRowWriter.sortField(field), buffer, 4, elementWidth(field));
+ final Field elementField = this.field.getChildren().get(0);
+ fixedWidth = fixedWidthFor(elementField);
+ elementNullable = elementField.isNullable();
+ }
+
+ public static int elementWidth(final Field field) {
+ final int width = fixedWidthFor(field.getChildren().get(0));
+ if (width < 0) {
+ return 8;
+ } else {
+ return width;
+ }
+ }
+
+ @Override
+ protected int writeNumElements() {
+ buffer.putInt32(startIndex, numElements);
+ return 4;
+ }
+
+ @Override
+ protected int calculateHeaderInBytes() {
+ return CompactBinaryArray.calculateHeaderInBytes(fixedWidth, numElements, elementNullable);
+ }
+
+ @Override
+ protected void primitiveArrayAdvance(final int size) {
+ buffer._increaseWriterIndexUnsafe(size);
+ }
+
+ @Override
+ protected int bufferWriteIndexFor(final int ordinal) {
+ if (fixedWidth > 0) {
+ return getOffset(ordinal);
+ } else {
+ return super.bufferWriteIndexFor(ordinal);
+ }
+ }
+
+ @Override
+ protected boolean copyShouldIncreaseWriterIndex(final int ordinal) {
+ if (fixedWidth > 0) {
+ return false;
+ } else {
+ return super.copyShouldIncreaseWriterIndex(ordinal);
+ }
+ }
+
+ @Override
+ public void setOffsetAndSize(final int ordinal, final int absoluteOffset, final int size) {
+ if (fixedWidth > 0) {
+ return;
+ }
+ super.setOffsetAndSize(ordinal, absoluteOffset, size);
+ }
+
+ @Override
+ public void setNotNullAt(final int ordinal) {
+ if (!elementNullable) {
+ return;
+ }
+ super.setNotNullAt(ordinal);
+ }
+
+ @Override
+ public void setNullAt(final int ordinal) {
+ if (!elementNullable) {
+ throw new NullPointerException("unexpected null element at ordinal " + ordinal);
+ }
+ super.setNullAt(ordinal);
+ if (fixedWidth > 0) {
+ final int off = getOffset(ordinal);
+ for (int i = off; i < off + fixedWidth; i++) {
+ getBuffer().putByte(i, 0);
+ }
+ }
+ }
+
+ public void resetFor(final CompactBinaryRowWriter nestedWriter, final int ordinal) {
+ if (fixedWidth > 0) {
+ nestedWriter.startIndex = getOffset(ordinal);
+ nestedWriter.resetHeader();
+ } else {
+ nestedWriter.reset();
+ }
+ }
+
+ @Override
+ public BinaryArray newArray() {
+ return new CompactBinaryArray(field);
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/CompactBinaryRowWriter.java b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/CompactBinaryRowWriter.java
new file mode 100644
index 000000000..aa92450e0
--- /dev/null
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/CompactBinaryRowWriter.java
@@ -0,0 +1,336 @@
+/*
+ * 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.fory.format.row.binary.writer;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.ArrowType.ArrowTypeID;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.fory.format.row.binary.BinaryRow;
+import org.apache.fory.format.row.binary.CompactBinaryRow;
+import org.apache.fory.format.type.DataTypes;
+import org.apache.fory.memory.MemoryBuffer;
+
+/** Writer class to produce {@link CompactBinaryRow}-formatted rows. */
+public class CompactBinaryRowWriter extends BaseBinaryRowWriter {
+
+ private final int headerSize;
+ private final boolean allFieldsFixedSize;
+ private final boolean allFieldsNotNullable;
+ private final int fixedSize;
+ private final int[] fixedOffsets;
+
+ public CompactBinaryRowWriter(final Schema schema) {
+ super(sortSchema(schema), computeFixedRegionSize(schema));
+ headerSize = headerBytes(schema);
+ allFieldsFixedSize = allFieldsFixedSize(schema);
+ fixedSize = computeAlignedFixedRegionSize();
+ fixedOffsets = fixedOffsets(schema);
+ allFieldsNotNullable = allNotNullable(schema.getFields());
+ }
+
+ public CompactBinaryRowWriter(final Schema schema, final BinaryWriter writer) {
+ super(sortSchema(schema), writer, computeFixedRegionSize(schema));
+ headerSize = headerBytes(schema);
+ allFieldsFixedSize = allFieldsFixedSize(schema);
+ fixedSize = computeAlignedFixedRegionSize();
+ fixedOffsets = fixedOffsets(schema);
+ allFieldsNotNullable = allNotNullable(schema.getFields());
+ }
+
+ public CompactBinaryRowWriter(final Schema schema, final MemoryBuffer buffer) {
+ super(sortSchema(schema), buffer, computeFixedRegionSize(schema));
+ headerSize = headerBytes(schema);
+ allFieldsFixedSize = allFieldsFixedSize(schema);
+ fixedSize = computeAlignedFixedRegionSize();
+ fixedOffsets = fixedOffsets(schema);
+ allFieldsNotNullable = allNotNullable(schema.getFields());
+ }
+
+ private static boolean allFieldsFixedSize(final Schema schema) {
+ for (final Field f : schema.getFields()) {
+ if (fixedWidthFor(f) == -1) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private int computeAlignedFixedRegionSize() {
+ final int totalUnalignedSize = headerSize + bytesBeforeBitMap;
+ if (allFieldsFixedSize) {
+ return totalUnalignedSize;
+ }
+ return roundNumberOfBytesToNearestWord(totalUnalignedSize);
+ }
+
+ // TODO: this should use StableValue once it's available
+ public static boolean allNotNullable(final List<Field> fields) {
+ for (final Field f : fields) {
+ if (f.isNullable()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ public static int headerBytes(final Schema schema) {
+ return headerBytes(schema.getFields());
+ }
+
+ private static int headerBytes(final List<Field> fields) {
+ if (allNotNullable(fields)) {
+ return 0;
+ }
+ return (fields.size() + 7) / 8;
+ }
+
+ public static Schema sortSchema(final Schema schema) {
+ return new Schema(sortFields(schema.getFields()), schema.getCustomMetadata());
+ }
+
+ @Override
+ protected int fixedSize() {
+ return fixedSize;
+ }
+
+ @Override
+ protected int headerInBytes() {
+ return headerSize;
+ }
+
+ @Override
+ public int getOffset(final int ordinal) {
+ return startIndex + fixedOffsets[ordinal];
+ }
+
+ private static int fixedBinarySize(final Field field) {
+ final ArrowType type = field.getType();
+ if (type.getTypeID() == ArrowTypeID.FixedSizeBinary) {
+ return ((ArrowType.FixedSizeBinary) type).getByteWidth();
+ }
+ return -1;
+ }
+
+ /** Total size of fixed region: fixed size inline values plus variable sized values' pointers. */
+ static int computeFixedRegionSize(final Schema schema) {
+ int fixedSize = 0;
+ for (final Field f : schema.getFields()) {
+ fixedSize += fixedRegionSpaceFor(f);
+ }
+ return fixedSize;
+ }
+
+ /** Number of bytes used for a field if fixed, -1 if variable sized. */
+ public static int fixedWidthFor(final Schema schema, final int ordinal) {
+ return fixedWidthFor(schema.getFields().get(ordinal));
+ }
+
+ /** Number of bytes used for a field if fixed, -1 if variable sized. */
+ public static int fixedWidthFor(final Field f) {
+ int fixedWidth = DataTypes.getTypeWidth(f.getType());
+ if (fixedWidth == -1) {
+ if (f.getType().getTypeID() == ArrowTypeID.Struct) {
+ int nestedWidth = headerBytes(f.getChildren());
+ for (final Field child : f.getChildren()) {
+ final int childWidth = fixedWidthFor(child);
+ if (childWidth == -1) {
+ return -1;
+ }
+ nestedWidth += childWidth;
+ }
+ return nestedWidth;
+ }
+ fixedWidth = fixedBinarySize(f);
+ }
+ return fixedWidth;
+ }
+
+ /**
+ * Number of bytes used in fixed-data area for a field - 8 (combined offset + size) if
+ * variable-sized.
+ */
+ static int fixedRegionSpaceFor(final Field f) {
+ final int fixedWidth = fixedWidthFor(f);
+ if (fixedWidth == -1) {
+ return 8;
+ }
+ return fixedWidth;
+ }
+
+ // TODO: this should use StableValue once it's available
+ public static int[] fixedOffsets(final Schema schema) {
+ final List<Field> fields = schema.getFields();
+ final int[] result = new int[fields.size() + 1];
+ int off = 0;
+ for (int i = 0; i < fields.size(); i++) {
+ result[i] = off;
+ off += fixedRegionSpaceFor(fields.get(i));
+ }
+ result[fields.size()] = off;
+ return result;
+ }
+
+ static List<Field> sortFields(final List<Field> fields) {
+ final ArrayList<Field> sortedFields = new ArrayList<>();
+ for (final Field f : fields) {
+ sortedFields.add(sortField(f));
+ }
+ Collections.sort(sortedFields, new FieldAlignmentComparator());
+ return sortedFields;
+ }
+
+ static Field sortField(final Field field) {
+ final List<Field> children = field.getChildren();
+ final List<Field> sortedChildren = sortFields(children);
+ if (children.equals(sortedChildren)) {
+ return field;
+ }
+ return new Field(field.getName(), field.getFieldType(), sortedChildren);
+ }
+
+ public void resetFor(final CompactBinaryRowWriter nestedWriter, final int ordinal) {
+ if (fixedWidthFor(getSchema(), ordinal) == -1) {
+ nestedWriter.reset();
+ } else {
+ nestedWriter.startIndex = getOffset(ordinal);
+ nestedWriter.resetHeader();
+ }
+ }
+
+ @Override
+ protected void resetHeader() {
+ final int bitmapStart = startIndex + bytesBeforeBitMap;
+ final int end = startIndex + computeAlignedFixedRegionSize();
+ for (int i = bitmapStart; i < end; i += 1) {
+ buffer.putByte(i, 0);
+ }
+ }
+
+ @Override
+ public void setNullAt(final int ordinal) {
+ if (allFieldsNotNullable) {
+ throw new IllegalStateException(
+ "Field " + getSchema().getFields().get(ordinal) + " has null value for not-null field");
+ }
+ super.setNullAt(ordinal);
+ }
+
+ @Override
+ protected void clearValue(final int ordinal) {
+ final int fixedWidth = fixedWidthFor(getSchema(), ordinal);
+ if (fixedWidth > 0) {
+ final int off = getOffset(ordinal);
+ for (int i = off; i < off + fixedWidth; i++) {
+ getBuffer().putByte(i, 0);
+ }
+ } else {
+ super.clearValue(ordinal);
+ }
+ }
+
+ @Override
+ public void setNotNullAt(final int ordinal) {
+ if (allFieldsNotNullable) {
+ return;
+ }
+ super.setNotNullAt(ordinal);
+ }
+
+ @Override
+ public boolean isNullAt(final int ordinal) {
+ if (allFieldsNotNullable) {
+ return false;
+ }
+ return super.isNullAt(ordinal);
+ }
+
+ @Override
+ public void write(final int ordinal, final byte value) {
+ final int offset = getOffset(ordinal);
+ buffer.putByte(offset, value);
+ }
+
+ @Override
+ public void write(final int ordinal, final boolean value) {
+ final int offset = getOffset(ordinal);
+ buffer.putBoolean(offset, value);
+ }
+
+ @Override
+ public void write(final int ordinal, final short value) {
+ final int offset = getOffset(ordinal);
+ buffer.putInt16(offset, value);
+ }
+
+ @Override
+ public void write(final int ordinal, final int value) {
+ final int offset = getOffset(ordinal);
+ buffer.putInt32(offset, value);
+ }
+
+ @Override
+ public void write(final int ordinal, final float value) {
+ final int offset = getOffset(ordinal);
+ buffer.putFloat32(offset, value);
+ }
+
+ @Override
+ public void writeUnaligned(
+ final int ordinal, final byte[] input, final int offset, final int numBytes) {
+ final int inlineWidth = fixedWidthFor(getSchema(), ordinal);
+ if (inlineWidth > 0) {
+ buffer.put(getOffset(ordinal), input, 0, numBytes);
+ } else {
+ super.writeUnaligned(ordinal, input, offset, numBytes);
+ }
+ }
+
+ @Override
+ public void writeUnaligned(
+ final int ordinal, final MemoryBuffer input, final int offset, final int numBytes) {
+ final int inlineWidth = fixedWidthFor(getSchema(), ordinal);
+ if (inlineWidth > 0) {
+ assert inlineWidth == numBytes;
+ buffer.copyFrom(getOffset(ordinal), input, 0, numBytes);
+ } else {
+ super.writeUnaligned(ordinal, input, offset, numBytes);
+ }
+ }
+
+ @Override
+ public void writeAlignedBytes(
+ final int ordinal, final MemoryBuffer input, final int baseOffset, final int numBytes) {
+ final int inlineWidth = fixedWidthFor(getSchema(), ordinal);
+ if (inlineWidth > 0) {
+ buffer.copyFrom(getOffset(ordinal), input, 0, numBytes);
+ } else {
+ super.writeAlignedBytes(ordinal, input, baseOffset, numBytes);
+ }
+ }
+
+ @Override
+ protected BinaryRow newRow() {
+ return new CompactBinaryRow(getSchema(), fixedOffsets);
+ }
+}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoder.java b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/FieldAlignmentComparator.java
similarity index 51%
copy from java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoder.java
copy to java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/FieldAlignmentComparator.java
index 3216b7ca1..651a05e93 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/encoder/Encoder.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/row/binary/writer/FieldAlignmentComparator.java
@@ -17,24 +17,26 @@
* under the License.
*/
-package org.apache.fory.format.encoder;
+package org.apache.fory.format.row.binary.writer;
-import org.apache.fory.memory.MemoryBuffer;
+import java.util.Comparator;
+import org.apache.arrow.vector.types.pojo.Field;
-/**
- * The encoding interface for encode/decode object to/from binary. The implementation class must
- * have a constructor with signature {@code Object[] references}, so we can pass any params to
- * codec.
- *
- * @param <T> type of value
- */
-public interface Encoder<T> {
-
- T decode(MemoryBuffer buffer);
-
- T decode(byte[] bytes);
-
- byte[] encode(T obj);
+class FieldAlignmentComparator implements Comparator<Field> {
+ @Override
+ public int compare(final Field f1, final Field f2) {
+ final int f1Size = CompactBinaryRowWriter.fixedRegionSpaceFor(f1);
+ final int f2Size = CompactBinaryRowWriter.fixedRegionSpaceFor(f2);
+ final int f1Align = tryAlign(f1Size);
+ final int f2Align = tryAlign(f2Size);
+ final int alignCmp = Integer.compare(f1Align, f2Align);
+ if (alignCmp != 0) {
+ return alignCmp;
+ }
+ return -Integer.compare(f1Size, f2Size);
+ }
- void encode(MemoryBuffer buffer, T obj);
+ protected int tryAlign(final int size) {
+ return size == 4 || size == 2 || (size & 7) == 0 ? 0 : 1;
+ }
}
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/type/DataTypes.java b/java/fory-format/src/main/java/org/apache/fory/format/type/DataTypes.java
index 216e2868e..a918d260c 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/type/DataTypes.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/type/DataTypes.java
@@ -74,6 +74,12 @@ public class DataTypes {
return -1;
}
+ @Override
+ public Integer visit(
+ org.apache.arrow.vector.types.pojo.ArrowType.FixedSizeBinary type) {
+ return type.getByteWidth();
+ }
+
@Override
public Integer visit(org.apache.arrow.vector.types.pojo.ArrowType.List type) {
return -1;
@@ -91,7 +97,7 @@ public class DataTypes {
@Override
public Integer visit(org.apache.arrow.vector.types.pojo.ArrowType.Int type) {
- return type.getBitWidth() / 8;
+ return (type.getBitWidth() + 7) / 8;
}
@Override
@@ -211,6 +217,12 @@ public class DataTypes {
return ArrowType.STRUCT;
}
+ @Override
+ public ArrowType visit(
+ org.apache.arrow.vector.types.pojo.ArrowType.FixedSizeBinary type) {
+ return ArrowType.FIXED_SIZE_BINARY;
+ }
+
@Override
public ArrowType visit(org.apache.arrow.vector.types.pojo.ArrowType.List type) {
return ArrowType.LIST;
diff --git a/java/fory-format/src/main/java/org/apache/fory/format/type/TypeInference.java b/java/fory-format/src/main/java/org/apache/fory/format/type/TypeInference.java
index 7c5294231..031aae881 100644
--- a/java/fory-format/src/main/java/org/apache/fory/format/type/TypeInference.java
+++ b/java/fory-format/src/main/java/org/apache/fory/format/type/TypeInference.java
@@ -163,6 +163,10 @@ public class TypeInference {
if (replacementField != null) {
return replacementField;
}
+ TypeRef<?> replacementType = customEncoder.encodedType();
+ if (replacementType != null && !typeRef.equals(replacementType)) {
+ return inferField(name, replacementType, ctx);
+ }
}
if (rawType == boolean.class) {
return field(name, DataTypes.notNullFieldType(ArrowType.Bool.INSTANCE));
diff --git a/java/fory-format/src/test/java/org/apache/fory/format/encoder/ArrayEncoderTest.java b/java/fory-format/src/test/java/org/apache/fory/format/encoder/ArrayEncoderTest.java
index c5f795c40..52d35ef38 100644
--- a/java/fory-format/src/test/java/org/apache/fory/format/encoder/ArrayEncoderTest.java
+++ b/java/fory-format/src/test/java/org/apache/fory/format/encoder/ArrayEncoderTest.java
@@ -44,7 +44,7 @@ public class ArrayEncoderTest {
}
ArrayEncoder<List<RowEncoderTest.Bar>> encoder =
- Encoders.arrayEncoder(bars.getClass(), RowEncoderTest.Bar.class);
+ Encoders.arrayEncoder(new TypeRef<List<RowEncoderTest.Bar>>() {});
BinaryArray array = encoder.toArray(bars);
List<RowEncoderTest.Bar> newBars = encoder.fromArray(array);
diff --git a/java/fory-format/src/test/java/org/apache/fory/format/encoder/CodecBuilderTest.java b/java/fory-format/src/test/java/org/apache/fory/format/encoder/CodecBuilderTest.java
index 3bb5be7cc..9c9dfb525 100644
--- a/java/fory-format/src/test/java/org/apache/fory/format/encoder/CodecBuilderTest.java
+++ b/java/fory-format/src/test/java/org/apache/fory/format/encoder/CodecBuilderTest.java
@@ -39,13 +39,14 @@ public class CodecBuilderTest {
@Test
public void loadOrGenRowCodecClass() {
- Class<?> codecClass = Encoders.loadOrGenRowCodecClass(BeanA.class);
+ Class<?> codecClass = Encoders.loadOrGenRowCodecClass(BeanA.class, DefaultCodecFormat.INSTANCE);
assertTrue(GeneratedRowEncoder.class.isAssignableFrom(codecClass));
assertTrue(
- GeneratedRowEncoder.class.isAssignableFrom(Encoders.loadOrGenRowCodecClass(BeanB.class)));
+ GeneratedRowEncoder.class.isAssignableFrom(
+ Encoders.loadOrGenRowCodecClass(BeanB.class, DefaultCodecFormat.INSTANCE)));
assertTrue(
GeneratedRowEncoder.class.isAssignableFrom(
- Encoders.loadOrGenRowCodecClass(AtomicLong.class)));
+ Encoders.loadOrGenRowCodecClass(AtomicLong.class, DefaultCodecFormat.INSTANCE)));
}
static void testStreamingEncode(Encoder encoder, Object object) {
diff --git a/java/fory-format/src/test/java/org/apache/fory/format/encoder/CompactCodecTest.java b/java/fory-format/src/test/java/org/apache/fory/format/encoder/CompactCodecTest.java
new file mode 100644
index 000000000..fd5849904
--- /dev/null
+++ b/java/fory-format/src/test/java/org/apache/fory/format/encoder/CompactCodecTest.java
@@ -0,0 +1,708 @@
+/*
+ * 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.fory.format.encoder;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotSame;
+
+import java.nio.ByteBuffer;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.Set;
+import java.util.UUID;
+import lombok.Data;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.fory.format.encoder.RowEncoderTest.Bar;
+import org.apache.fory.format.row.binary.BinaryArray;
+import org.apache.fory.format.row.binary.BinaryMap;
+import org.apache.fory.format.row.binary.BinaryRow;
+import org.apache.fory.format.row.binary.CompactBinaryRow;
+import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.memory.MemoryUtils;
+import org.apache.fory.reflect.TypeRef;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class CompactCodecTest {
+
+ static {
+ Encoders.registerCustomCodec(UUID.class, new CompactUUIDCodec());
+ Encoders.registerCustomCodec(NotNullByte.class, new NotNullByteCodec());
+ }
+
+ @Data
+ public static class CompactType {
+ public float f1;
+ public double f2;
+ public byte f3;
+ public short f4;
+ public int f5;
+ public long f6;
+ public String f7;
+ }
+
+ @Test
+ public void testCompactType() {
+ final CompactType bean1 = new CompactType();
+ bean1.f1 = 1;
+ bean1.f2 = 2;
+ bean1.f3 = 3;
+ bean1.f4 = 4;
+ bean1.f5 = 5;
+ bean1.f6 = 6;
+ bean1.f7 = "7";
+ final RowEncoder<CompactType> encoder =
+ Encoders.buildBeanCodec(CompactType.class).compactEncoding().build().get();
+ final List<Field> fields = encoder.schema().getFields();
+ assertEquals(fields.get(0).getName(), "f2");
+ assertEquals(fields.get(1).getName(), "f6");
+ assertEquals(fields.get(2).getName(), "f7");
+ assertEquals(fields.get(3).getName(), "f1");
+ assertEquals(fields.get(4).getName(), "f5");
+ assertEquals(fields.get(5).getName(), "f4");
+ assertEquals(fields.get(6).getName(), "f3");
+
+ final BinaryRow row = encoder.toRow(bean1);
+ assertEquals(row.getClass(), CompactBinaryRow.class);
+ assertEquals(row.getOffset(0), 0);
+ assertEquals(row.getFloat64(0), 2);
+ final MemoryBuffer buffer = MemoryUtils.wrap(row.toBytes());
+ row.pointTo(buffer, 0, buffer.size());
+ final CompactType deserializedBean = encoder.fromRow(row);
+ assertEquals(bean1, deserializedBean);
+ assertEquals(buffer.size(), 4 + 7 + 1 + 4 + 8 + 1 + 2 + 4 + 8 + 8 + 1);
+ }
+
+ @Test
+ public void testEncodeResetsNullBits() {
+ final CompactType bean1 = new CompactType();
+ bean1.f1 = 1;
+ bean1.f2 = 2;
+ bean1.f3 = 3;
+ bean1.f4 = 4;
+ bean1.f5 = 5;
+ bean1.f6 = 6;
+ final RowEncoder<CompactType> encoder =
+ Encoders.buildBeanCodec(CompactType.class).compactEncoding().build().get();
+
+ byte[] encoded = encoder.encode(bean1);
+ bean1.f7 = "Not null!";
+ encoded = encoder.encode(bean1);
+ final CompactType deserializedBean = encoder.decode(encoded);
+ assertEquals(deserializedBean, bean1);
+ }
+
+ @Data
+ public static class CompactListType {
+ public List<CompactType> f1;
+ }
+
+ @Test
+ public void testCompactListType() {
+ final CompactType bean1 = new CompactType();
+ bean1.f1 = 1;
+ bean1.f2 = 2;
+ bean1.f3 = 3;
+ bean1.f4 = 4;
+ bean1.f5 = 5;
+ bean1.f6 = 6;
+ bean1.f7 = "7";
+
+ final CompactType bean2 = new CompactType();
+ bean2.f1 = 7;
+ bean2.f2 = 6;
+ bean2.f3 = 5;
+ bean2.f4 = 4;
+ bean2.f5 = 3;
+ bean2.f6 = 2;
+ bean2.f7 = "1";
+
+ final CompactListType list = new CompactListType();
+ list.f1 = Arrays.asList(bean1, bean2);
+
+ final RowEncoder<CompactListType> encoder =
+ Encoders.buildBeanCodec(CompactListType.class).compactEncoding().build().get();
+
+ final BinaryRow row = encoder.toRow(list);
+ assertEquals(row.getClass(), CompactBinaryRow.class);
+ assertEquals(row.getArray(0).getStruct(0).getClass(), CompactBinaryRow.class);
+ final MemoryBuffer buffer = MemoryUtils.wrap(row.toBytes());
+ row.pointTo(buffer, 0, buffer.size());
+ final CompactListType deserializedBean = encoder.fromRow(row);
+ assertEquals(deserializedBean, list);
+ assertEquals(buffer.size(), 37 + 48 + 48);
+ }
+
+ @Data
+ public static class CompactUuidType {
+ public UUID f1;
+
+ public CompactUuidType() {}
+
+ public CompactUuidType(final UUID f1) {
+ this.f1 = f1;
+ }
+ }
+
+ @Test
+ public void testCompactUuidType() {
+ final CompactUuidType bean1 = new CompactUuidType();
+ bean1.f1 = new UUID(42, 24);
+ final RowEncoder<CompactUuidType> encoder =
+ Encoders.buildBeanCodec(CompactUuidType.class).compactEncoding().build().get();
+ final BinaryRow row = encoder.toRow(bean1);
+ final MemoryBuffer buffer = MemoryUtils.wrap(row.toBytes());
+ row.pointTo(buffer, 0, buffer.size());
+ final CompactUuidType deserializedBean = encoder.fromRow(row);
+ assertEquals(bean1, deserializedBean);
+ assertEquals(buffer.size(), 16 + 1);
+ }
+
+ static class CompactUUIDCodec implements CustomCodec.MemoryBufferCodec<UUID> {
+ @Override
+ public Field getField(final String fieldName) {
+ return Field.nullable(fieldName, new ArrowType.FixedSizeBinary(16));
+ }
+
+ @Override
+ public MemoryBuffer encode(final UUID value) {
+ final MemoryBuffer result = MemoryBuffer.newHeapBuffer(16);
+ result.putInt64(0, value.getMostSignificantBits());
+ result.putInt64(8, value.getLeastSignificantBits());
+ return result;
+ }
+
+ @Override
+ public UUID decode(final MemoryBuffer value) {
+ return new UUID(value.readInt64(), value.readInt64());
+ }
+ }
+
+ @Data
+ public static class Nested1 {
+ public short f1;
+ }
+
+ @Data
+ public static class Nested2 {
+ public int f1;
+ }
+
+ @Test
+ public void testAllNonnullElideBitmap() {
+ final Nested1 bean1 = new Nested1();
+ bean1.f1 = 42;
+ final Nested2 bean2 = new Nested2();
+ bean2.f1 = 75;
+ final RowEncoder<Nested1> encoder =
+ Encoders.buildBeanCodec(Nested1.class).compactEncoding().build().get();
+ BinaryRow row = encoder.toRow(bean1);
+ MemoryBuffer buffer = MemoryUtils.wrap(row.toBytes());
+ row.pointTo(buffer, 0, buffer.size());
+ final Nested1 deserializedBean = encoder.fromRow(row);
+ assertEquals(bean1, deserializedBean);
+ assertEquals(buffer.size(), 2);
+
+ final RowEncoder<Nested2> encoder2 =
+ Encoders.buildBeanCodec(Nested2.class).compactEncoding().build().get();
+ row = encoder2.toRow(bean2);
+ buffer = MemoryUtils.wrap(row.toBytes());
+ row.pointTo(buffer, 0, buffer.size());
+ final Nested2 deserializedBean2 = encoder2.fromRow(row);
+ assertEquals(bean2, deserializedBean2);
+ assertEquals(buffer.size(), 4);
+ }
+
+ @Data
+ public static class InlineNestedType {
+ public Nested1 f1;
+ public Nested2 f2;
+ }
+
+ @Test
+ public void testInlineNestedType() {
+ final InlineNestedType bean1 = new InlineNestedType();
+ bean1.f1 = new Nested1();
+ bean1.f1.f1 = 42;
+ bean1.f2 = new Nested2();
+ bean1.f2.f1 = 75;
+ final RowEncoder<InlineNestedType> encoder =
+ Encoders.buildBeanCodec(InlineNestedType.class).compactEncoding().build().get();
+ final BinaryRow row = encoder.toRow(bean1);
+ assertEquals(row.getSchema().getFields().get(0).getName(), "f2");
+ assertEquals(row.getSchema().getFields().get(1).getName(), "f1");
+ assertEquals(row.getOffset(0), 0);
+ assertEquals(row.getOffset(1), 4);
+ final MemoryBuffer buffer = MemoryUtils.wrap(row.toBytes());
+ row.pointTo(buffer, 0, buffer.size());
+ final InlineNestedType deserializedBean = encoder.fromRow(row);
+ assertEquals(deserializedBean, bean1);
+ assertEquals(buffer.size(), 7);
+ }
+
+ @Data
+ public static class InlineNestedArrayType {
+ public InlineNestedType f1;
+ public String f2;
+ public UUID[] f3;
+ }
+
+ @Test
+ public void testInlineNestedArrayType() {
+ final InlineNestedArrayType bean1 = new InlineNestedArrayType();
+ bean1.f1 = new InlineNestedType();
+ bean1.f1.f1 = new Nested1();
+ bean1.f1.f1.f1 = 42;
+ bean1.f1.f2 = new Nested2();
+ bean1.f1.f2.f1 = 75;
+ bean1.f2 = "luna";
+ bean1.f3 = new UUID[] {new UUID(1, 2), new UUID(3, 4), new UUID(5, 6)};
+ final RowEncoder<InlineNestedArrayType> encoder =
+ Encoders.buildBeanCodec(InlineNestedArrayType.class).compactEncoding().build().get();
+ final BinaryRow row = encoder.toRow(bean1);
+ final MemoryBuffer buffer = MemoryUtils.wrap(row.toBytes());
+ row.pointTo(buffer, 0, buffer.size());
+ final InlineNestedArrayType deserializedBean = encoder.fromRow(row);
+ assertEquals(deserializedBean, bean1);
+ assertEquals(buffer.size(), 88);
+ }
+
+ @Data
+ public static class InlinePrimitiveNestedArrayType {
+ public InlineNestedType f1;
+ public String f2;
+ public short[] f3;
+ }
+
+ @Test
+ public void testInlinePrimitiveNestedArrayType() {
+ final InlinePrimitiveNestedArrayType bean1 = new InlinePrimitiveNestedArrayType();
+ bean1.f1 = new InlineNestedType();
+ bean1.f1.f1 = new Nested1();
+ bean1.f1.f1.f1 = 42;
+ bean1.f1.f2 = new Nested2();
+ bean1.f1.f2.f1 = 75;
+ bean1.f2 = "luna";
+ bean1.f3 = new short[] {1, 2, 3};
+ final RowEncoder<InlinePrimitiveNestedArrayType> encoder =
+ Encoders.buildBeanCodec(InlinePrimitiveNestedArrayType.class)
+ .compactEncoding()
+ .build()
+ .get();
+ final BinaryRow row = encoder.toRow(bean1);
+ final MemoryBuffer buffer = MemoryUtils.wrap(row.toBytes());
+ row.pointTo(buffer, 0, buffer.size());
+ final InlinePrimitiveNestedArrayType deserializedBean = encoder.fromRow(row);
+ assertEquals(deserializedBean, bean1);
+ assertEquals(buffer.size(), 50);
+ }
+
+ @Data
+ public static class CompactMapType {
+ public Map<UUID, CompactUuidType> map;
+ }
+
+ @Test
+ public void testCompactMapType() {
+ final CompactMapType bean1 = new CompactMapType();
+ bean1.map = new HashMap<>();
+ final UUID u1 = new UUID(42, 24);
+ bean1.map.put(u1, new CompactUuidType(u1));
+ final UUID u2 = new UUID(55, 66);
+ bean1.map.put(u2, new CompactUuidType(u2));
+ final RowEncoder<CompactMapType> encoder =
+ Encoders.buildBeanCodec(CompactMapType.class).compactEncoding().build().get();
+ final BinaryRow row = encoder.toRow(bean1);
+ assertEquals(
+ row.getMap(0).keyArray().toString(),
+ "[0x2a000000000000001800000000000000,0x37000000000000004200000000000000]");
+ assertEquals(
+ row.getMap(0).valueArray().toString(),
+ "[{f1=0x2a000000000000001800000000000000},{f1=0x37000000000000004200000000000000}]");
+ final MemoryBuffer buffer = MemoryUtils.wrap(row.toBytes());
+ row.pointTo(buffer, 0, buffer.size());
+ final CompactMapType deserializedBean = encoder.fromRow(row);
+ assertEquals(deserializedBean, bean1);
+ assertEquals(buffer.size(), 109);
+ }
+
+ @Test
+ public void testTwoFieldsAndASet() {
+ final TwoFieldsAndASet bean1 = new TwoFieldsAndASet();
+ bean1.f1 = new UUID(42, 24);
+ bean1.f2 = new UUID(55, 66);
+ bean1.f3 = Set.of(LocalDate.of(2112, 1, 1));
+ final RowEncoder<TwoFieldsAndASet> encoder =
+ Encoders.buildBeanCodec(TwoFieldsAndASet.class).compactEncoding().build().get();
+ final BinaryRow row = encoder.toRow(bean1);
+ final MemoryBuffer buffer = MemoryUtils.wrap(row.toBytes());
+ row.pointTo(buffer, 0, buffer.size());
+ final TwoFieldsAndASet deserializedBean = encoder.fromRow(row);
+ assertEquals(deserializedBean, bean1);
+ }
+
+ @Data
+ public static class TwoFieldsAndASet {
+ public UUID f1;
+ public UUID f2;
+ public Set<LocalDate> f3;
+ }
+
+ @Test
+ public void testNestedConfigObject() {
+ final InnerConfigObject inner = new InnerConfigObject();
+ inner.f1 = new UUID(1, 2);
+ inner.f2 = new UUID(3, 4);
+ inner.f3 = new UUID(5, 6);
+ inner.f4 = true;
+ inner.f5 = "Indubitably";
+ inner.f6 = EnumSet.of(ConfigEnum.B, ConfigEnum.C);
+ final OuterConfigObject bean1 = new OuterConfigObject();
+ bean1.f1 = inner;
+ bean1.f2 = LocalDate.of(2112, 1, 1);
+ final RowEncoder<OuterConfigObject> encoder =
+ Encoders.buildBeanCodec(OuterConfigObject.class).compactEncoding().build().get();
+ final BinaryRow row = encoder.toRow(bean1);
+ final MemoryBuffer buffer = MemoryUtils.wrap(row.toBytes());
+ row.pointTo(buffer, 0, buffer.size());
+ final OuterConfigObject deserializedBean = encoder.fromRow(row);
+ assertEquals(deserializedBean, bean1);
+ }
+
+ @Data
+ public static class OuterConfigObject {
+ public InnerConfigObject f1;
+ public LocalDate f2;
+ }
+
+ @Data
+ public static class InnerConfigObject {
+ public UUID f1;
+ public UUID f2;
+ public UUID f3;
+ public boolean f4;
+ public String f5;
+ public Set<ConfigEnum> f6;
+ }
+
+ public enum ConfigEnum {
+ A,
+ B,
+ C
+ }
+
+ @Test
+ public void testBiglyBean() {
+ final BiglyBean big = new BiglyBean();
+ big.f1 = new UUID(1, 2);
+ big.f2 = ConfigEnum.B;
+ big.f3 = ConfigEnum.C;
+ big.f4 = new UUID(3, 4);
+ big.f5 = new UUID(5, 6);
+ big.f6 = Optional.of("Indubitably");
+ big.f7 = LocalDate.of(2112, 2, 2);
+ big.f8 = Optional.of(LocalDate.of(1221, 3, 4));
+ big.f9 = OptionalLong.of(-42);
+ big.f10 = OptionalLong.of(-24);
+ big.f11 = 1234;
+ big.f12 = 4321;
+ big.f13 = Instant.ofEpochMilli(12345678);
+ big.f14 = Instant.ofEpochMilli(87654321);
+ big.f15 = Optional.of(ConfigEnum.B);
+ final RowEncoder<BiglyBean> encoder =
+ Encoders.buildBeanCodec(BiglyBean.class).compactEncoding().build().get();
+ final BinaryRow row = encoder.toRow(big);
+ final MemoryBuffer buffer = MemoryUtils.wrap(row.toBytes());
+ row.pointTo(buffer, 0, buffer.size());
+ final BiglyBean deserializedBean = encoder.fromRow(row);
+ assertEquals(deserializedBean, big);
+ }
+
+ @Data
+ public static class BiglyBean {
+ public UUID f1;
+ public ConfigEnum f2;
+ public ConfigEnum f3;
+ public UUID f4;
+ public UUID f5;
+ public Optional<String> f6;
+ public LocalDate f7;
+ public Optional<LocalDate> f8;
+ public OptionalLong f9;
+ public OptionalLong f10;
+ public long f11;
+ public long f12;
+ public Instant f13;
+ public Instant f14;
+ public Optional<ConfigEnum> f15;
+ }
+
+ @Test
+ public void testArrayOfStruct() {
+ final ArrayNestedBean bean = new ArrayNestedBean();
+ bean.f1 = new UUID(1, 2);
+ bean.f2 = new UUID(3, 4);
+ bean.f3 = Instant.ofEpochMilli(12345678);
+ bean.f4 = true;
+ bean.f5 = Instant.ofEpochMilli(87654321);
+ bean.f6 = ConfigEnum.B;
+ bean.f7 = new UUID(5, 6);
+ bean.f8 = false;
+ bean.f9 = new ArrayNestedNestedBean();
+ bean.f9.f1 = Instant.ofEpochMilli(43218765);
+ bean.f9.f2 = "Indubitably";
+ bean.f9.f3 = EnumSet.of(ConfigEnum.B, ConfigEnum.C);
+ bean.f9.f4 = new UUID(7, 8);
+ bean.f9.f5 = "luna";
+ bean.f9.f6 = List.of();
+ final ArrayEncoder<List<ArrayNestedBean>> encoder =
+ Encoders.buildArrayCodec(new TypeRef<List<ArrayNestedBean>>() {})
+ .compactEncoding()
+ .build()
+ .get();
+ final List<ArrayNestedBean> expected = List.of(bean, bean);
+ final BinaryArray arr = encoder.toArray(expected);
+ final MemoryBuffer buffer = MemoryUtils.wrap(arr.toBytes());
+ arr.pointTo(buffer, 0, buffer.size());
+ final List<ArrayNestedBean> deserializedBean = encoder.fromArray(arr);
+ assertEquals(deserializedBean, expected);
+ }
+
+ @Data
+ public static class ArrayNestedBean {
+ public UUID f1;
+ public UUID f2;
+ public Instant f3;
+ public boolean f4;
+ public Instant f5;
+ public ConfigEnum f6;
+ public UUID f7;
+ public boolean f8;
+ public ArrayNestedNestedBean f9;
+ }
+
+ @Data
+ public static class ArrayNestedNestedBean {
+ public Instant f1;
+ public String f2;
+ public Set<ConfigEnum> f3;
+ public UUID f4;
+ public String f5;
+ public List<ArrayNestedNestedNestedBean> f6;
+ }
+
+ @Data
+ public static class ArrayNestedNestedNestedBean {
+ public Instant f1;
+ public String f2;
+ public String f3;
+ public String f4;
+ }
+
+ @Test
+ public void testNotNullByteArray() {
+ final List<NotNullByte> expected = new ArrayList<>();
+ for (int i = 0; i < 64; i++) {
+ expected.add(new NotNullByte((byte) i));
+ }
+ final ArrayEncoder<List<NotNullByte>> encoder =
+ Encoders.buildArrayCodec(new TypeRef<List<NotNullByte>>() {})
+ .compactEncoding()
+ .build()
+ .get();
+ final BinaryArray arr = encoder.toArray(expected);
+ final MemoryBuffer buffer = MemoryUtils.wrap(arr.toBytes());
+ arr.pointTo(buffer, 0, buffer.size());
+ final List<NotNullByte> deserializedBean = encoder.fromArray(arr);
+ assertEquals(deserializedBean, expected);
+ assertEquals(arr.getSizeInBytes(), 68);
+ }
+
+ @Test
+ public void testArrayBufferReuse() {
+ final List<NotNullByte> expected = new ArrayList<>();
+ final List<NotNullByte> expectedRev = new ArrayList<>();
+ for (int i = 0; i < 128; i++) {
+ final NotNullByte val = new NotNullByte((byte) i);
+ expected.add(val);
+ if (i < 16) {
+ expectedRev.add(0, val);
+ }
+ }
+ final ArrayEncoder<List<NotNullByte>> encoder =
+ Encoders.buildArrayCodec(new TypeRef<List<NotNullByte>>() {})
+ .compactEncoding()
+ .build()
+ .get();
+ final BinaryArray arr = encoder.toArray(expected);
+ final BinaryArray subRev = encoder.toArray(expectedRev);
+ assertNotSame(arr.getBuffer(), subRev.getBuffer());
+ }
+
+ @Data
+ public static class NotNullByte {
+ public byte b;
+
+ public NotNullByte(final byte b) {
+ this.b = b;
+ }
+ }
+
+ public static class NotNullByteCodec implements CustomCodec<NotNullByte, Byte> {
+ @Override
+ public Field getField(final String fieldName) {
+ return Field.notNullable(fieldName, new ArrowType.Int(8, true));
+ }
+
+ @Override
+ public TypeRef<Byte> encodedType() {
+ return new TypeRef<Byte>() {};
+ }
+
+ @Override
+ public Byte encode(final NotNullByte value) {
+ return value.b;
+ }
+
+ @Override
+ public NotNullByte decode(final Byte value) {
+ return new NotNullByte(value);
+ }
+ }
+
+ @Test
+ public void testCompactMapEncoder() {
+ final InnerConfigObject inner = new InnerConfigObject();
+ inner.f1 = new UUID(1, 2);
+ inner.f2 = new UUID(3, 4);
+ inner.f3 = new UUID(5, 6);
+ inner.f4 = true;
+ inner.f5 = "Indubitably";
+ inner.f6 = EnumSet.of(ConfigEnum.B, ConfigEnum.C);
+ final MapEncoder<Map<UUID, InnerConfigObject>> encoder =
+ Encoders.buildMapCodec(new TypeRef<Map<UUID, InnerConfigObject>>() {})
+ .compactEncoding()
+ .build()
+ .get();
+ final var expected = Map.of(inner.f1, inner);
+ final BinaryMap map = encoder.toMap(expected);
+ final MemoryBuffer buffer = MemoryUtils.wrap(map.toBytes());
+ map.pointTo(buffer, 0, buffer.size());
+ final var deserializedMap = encoder.fromMap(map);
+ assertEquals(deserializedMap, expected);
+ }
+
+ @Test
+ public void testNullClearOffset() {
+ final RowEncoder<Bar> encoder =
+ Encoders.buildBeanCodec(Bar.class).compactEncoding().build().get();
+ final Bar bar = new Bar();
+ bar.f1 = 42;
+ bar.f2 = null;
+ final byte[] nullBefore = encoder.encode(bar);
+
+ bar.f2 = "not null";
+ // write offset and size
+ encoder.encode(bar);
+
+ bar.f2 = null;
+ final byte[] nullAfter = encoder.encode(bar);
+ Assert.assertEquals(nullAfter, nullBefore);
+ }
+
+ @Data
+ public static class MediumBean {
+ public UUID f1;
+ public LocalDate f2;
+ public UUID f3;
+ }
+
+ @Test
+ public void testOffsetBuffer() {
+ final MediumBean bean1 = new MediumBean();
+ bean1.f1 = new UUID(1, 2);
+ bean1.f2 = LocalDate.ofEpochDay(56);
+ bean1.f3 = new UUID(3, 4);
+
+ final MediumBean bean2 = new MediumBean();
+ bean2.f1 = new UUID(7, 8);
+ bean2.f2 = LocalDate.ofEpochDay(1112);
+ bean2.f3 = new UUID(9, 10);
+
+ final MediumBean bean3 = new MediumBean();
+ bean3.f1 = new UUID(13, 14);
+ bean3.f2 = LocalDate.ofEpochDay(1718);
+ bean3.f3 = new UUID(15, 16);
+
+ final ArrayEncoder<List<MediumBean>> encoder =
+ Encoders.buildArrayCodec(new TypeRef<List<MediumBean>>() {})
+ .compactEncoding()
+ .withSizeEmbedded(false)
+ .build()
+ .get();
+ final List<MediumBean> expected = Arrays.asList(bean1, bean2, bean3, null);
+ final byte[] bytes = encoder.encode(expected);
+ final int offset = 17;
+ final var buf = ByteBuffer.allocate(bytes.length + offset);
+ buf.put(new byte[offset]);
+ buf.put(bytes);
+ buf.position(offset);
+ buf.limit(offset + bytes.length);
+ final List<MediumBean> deserializedBean = encoder.decode(MemoryBuffer.fromByteBuffer(buf));
+ assertEquals(deserializedBean, expected);
+ }
+
+ @Data
+ public static class HeaderClearObject {
+ public UUID f1;
+ public InnerHeaderClearObject f2;
+ }
+
+ @Data
+ public static class InnerHeaderClearObject {
+ public UUID f1;
+ public UUID f2;
+ public UUID f3;
+ }
+
+ @Test
+ public void testHeaderClear() {
+ final InnerHeaderClearObject inner = new InnerHeaderClearObject();
+ inner.f1 = UUID.randomUUID();
+
+ final HeaderClearObject outer = new HeaderClearObject();
+ outer.f2 = inner;
+
+ final RowEncoder<HeaderClearObject> encoder =
+ Encoders.buildBeanCodec(HeaderClearObject.class).compactEncoding().build().get();
+ final byte[] bytes = encoder.encode(outer);
+ final HeaderClearObject deserializedBean = encoder.decode(bytes);
+ assertEquals(deserializedBean, outer);
+ }
+}
diff --git a/java/fory-format/src/test/java/org/apache/fory/format/encoder/CustomCodecTest.java b/java/fory-format/src/test/java/org/apache/fory/format/encoder/CustomCodecTest.java
index 33b0faa4b..c59b63f52 100644
--- a/java/fory-format/src/test/java/org/apache/fory/format/encoder/CustomCodecTest.java
+++ b/java/fory-format/src/test/java/org/apache/fory/format/encoder/CustomCodecTest.java
@@ -196,7 +196,7 @@ public class CustomCodecTest {
}
}
- static class UuidEncoder implements CustomCodec.MemoryBufferCodec<UUID> {
+ public static class UuidEncoder implements CustomCodec.MemoryBufferCodec<UUID> {
@Override
public MemoryBuffer encode(final UUID value) {
final MemoryBuffer result = MemoryBuffer.newHeapBuffer(16);
@@ -209,6 +209,11 @@ public class CustomCodecTest {
public UUID decode(final MemoryBuffer value) {
return new UUID(value.readInt64(), value.readInt64());
}
+
+ @Override
+ public Field getField(final String fieldName) {
+ return Field.nullable(fieldName, new ArrowType.FixedSizeBinary(16));
+ }
}
static class SortedSetOfUuidDecoder implements CustomCollectionFactory<UUID, SortedSet<UUID>> {
diff --git a/java/fory-format/src/test/java/org/apache/fory/format/encoder/RowEncoderTest.java b/java/fory-format/src/test/java/org/apache/fory/format/encoder/RowEncoderTest.java
index c80febb64..83774ea4f 100644
--- a/java/fory-format/src/test/java/org/apache/fory/format/encoder/RowEncoderTest.java
+++ b/java/fory-format/src/test/java/org/apache/fory/format/encoder/RowEncoderTest.java
@@ -114,7 +114,8 @@ public class RowEncoderTest {
PrivateStruct s = new PrivateStruct();
s.f1 = ofHashMap(10L, 100L);
s.f2 = ofHashMap("k", "v");
- PrivateStruct s1 = encoder.decode(encoder.encode(s));
+ byte[] encoded = encoder.encode(s);
+ PrivateStruct s1 = encoder.decode(encoded);
Assert.assertEquals(s1.f1, s.f1);
Assert.assertEquals(s1.f2, s.f2);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@fory.apache.org
For additional commands, e-mail: commits-help@fory.apache.org