feat(java): introduce new 'compact' row encoding

compact encoding will store fixed sized fields inline,
relaxes alignment considerations to preserve alignment where possible
but using space more effectively

Author: stevenschlansker

java/fory-core/src/main/java/org/apache/fory/builder/CodecBuilder.java

@@ -122,6 +122,8 @@ public CodecBuilder(CodegenContext ctx, TypeRef<?> beanType) {
         .forEach(ctx::reserveName);
   }
 
+  public abstract String codecClassName(Class<?> cls);
+
   /** Generate codec class code. */
   public abstract String genCode();
 

java/fory-core/src/main/java/org/apache/fory/memory/MemoryBuffer.java

@@ -40,7 +40,7 @@
  *   <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>
  *

java/fory-core/src/main/java/org/apache/fory/util/ExceptionUtils.java

@@ -71,4 +71,13 @@ public static RuntimeException handleReadFailed(Fory fory, Throwable t) {
   }
 
   public static void ignore(Object... args) {}
+
+  public static RuntimeException throwAnyway(Throwable t) {
+    throw throwEvadingChecks(t);
+  }
+
+  @SuppressWarnings({"unchecked", "TypeParameterUnusedInFormals"})
+  private static <E extends Throwable> E throwEvadingChecks(Throwable throwable) throws E {
+    throw (E) throwable;
+  }
 }

java/fory-format/src/main/java/org/apache/fory/format/encoder/ArrayCodecBuilder.java

@@ -0,0 +1,121 @@
+/*
+ * 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 BaseArrayEncoder<>(writer, generatedEncoderFactory.apply(writer));
+      }
+    };
+  }
+
+  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);
+        }
+      }
+    };
+  }
+}

java/fory-format/src/main/java/org/apache/fory/format/encoder/ArrayEncoderBuilder.java

@@ -31,7 +31,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.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 String genCode() {
             "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 String genCode() {
   @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 Expression buildEncodeExpression() {
 
     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);
 

java/fory-format/src/main/java/org/apache/fory/format/encoder/BaseArrayEncoder.java

@@ -0,0 +1,112 @@
+/*
+ * 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 BaseArrayEncoder<T> implements ArrayEncoder<T> {
+  private final BinaryArrayWriter writer;
+  private final GeneratedArrayEncoder codec;
+
+  BaseArrayEncoder(final BinaryArrayWriter writer, final GeneratedArrayEncoder codec) {
+    this.writer = writer;
+    this.codec = codec;
+  }
+
+  @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, buffer.readInt32());
+  }
+
+  @Override
+  public T decodeRemaining(final MemoryBuffer buffer) {
+    return decode(buffer, 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 void encode(final MemoryBuffer buffer, final T obj) {
+    final MemoryBuffer prevBuffer = writer.getBuffer();
+    final int writerIndex = buffer.writerIndex();
+    buffer.writeInt32(-1);
+    try {
+      writer.setBuffer(buffer);
+      final BinaryArray array = toArray(obj);
+      final int size = buffer.writerIndex() - writerIndex - 4;
+      assert size == array.getSizeInBytes();
+      buffer.putInt32(writerIndex, size);
+    } finally {
+      writer.setBuffer(prevBuffer);
+    }
+  }
+
+  @Override
+  public int bareEncode(final MemoryBuffer buffer, final T obj) {
+    final MemoryBuffer prevBuffer = writer.getBuffer();
+    final int writerIndex = buffer.writerIndex();
+    try {
+      writer.setBuffer(buffer);
+      final BinaryArray array = toArray(obj);
+      final int size = buffer.writerIndex() - writerIndex;
+      assert size == array.getSizeInBytes();
+      return size;
+    } finally {
+      writer.setBuffer(prevBuffer);
+    }
+  }
+}