perf(format): cache compact row layout per nested slot (#3717)

## Why?

Avoid redundant re-computation of compact codec offsets layout, saving
cpu and memory allocations

and

fix(format): honor input offset when copying inline-struct bytes

## What does this PR do?

    perf(format): cache compact row layout per nested slot
    
    Hoist offsets, widths, and nullability into CompactRowLayout, built
    once per encoder. getStruct does no per-call lookup; inline-struct
    reads go directly to the parent buffer; the writer reads schema-
    derived state from the cache; compact rows skip the unused extData[]
    allocation. For List<Struct> fields the parent layout also carries
    the element layout so getArray(i) does not rebuild it.
    
    Fix a latent inline-width bug in CompactBinaryRowWriter that was
    masked by the prior slicing getStruct: writeUnaligned and
    writeAlignedBytes dropped the input offset, so re-emitting a child
    row via BinaryWriter.write(int, BinaryRow) copied the wrong bytes.

## Related issues

## AI Contribution Checklist

- [x] Substantial AI assistance was used in this PR: `yes` / `no`

AI Usage Disclosure
- substantial_ai_assistance: yes
- scope: all
- affected_files_or_subsystems: Java row format
- ai_review: ✔️ 
- ai_review_artifacts: 

> One commit, perf(format): cache compact row layout per nested slot.
Caches fixedOffsets, per-field fixedWidths, allFieldsNotNullable, and
bitmapWidthInBytes in a new CompactRowLayout value, computed once per
array (element type is fixed) or memoized per parent extData slot.
> The getStruct hot path previously rebuilt fixedOffsets and re-walked
allNotNullable on every call.
> Recommendation: Approve. Numbers in the commit message match a real
bottleneck (the int[] + allNotNullable walk per nested decode), the
implementation preserves the existing wire behavior and ownership model,
no public API regression, all format tests pass, and the new caching
follows the existing extData-slot pattern rather than inventing a new
mechanism.
>  — Reviewed by Claude (Opus 4.7), 2026-05-29

> Skill used: I ran this through the fory-code-review skill, which
delegated the diff inspection to a read-only review subagent loaded with
the Fory review checklist, the lesson-derived red flags, and the Java
runtime rules.
> Verdict
> No blockers. Recommend submission. Both commits are coherent and
tested:
> - d5cbdc2 is a real, narrow bugfix: the three put/copyFrom(..., 0,
numBytes) sites in CompactBinaryRowWriter were copying from the start of
the source buffer instead of from baseOffset. After schema sorting,
nested inline structs land at non-zero offsets, so this corrupted writes
for any non-trivial schema.
CompactCodecTest.testCopyInlineStructViaWriter exercises that exact
failure mode. Wire format unchanged — only the source byte range was
wrong.
> - 2b1825d precomputes a CompactRowLayout once per writer instead of
recomputing field offsets and nested layouts on every write. Cache is
per-CompactBinaryRowWriter instance, no map, no growth, no thread-safety
surface (writers are already not thread-safe). CompactRowLayoutTest pins
identity via assertSame at multiple nesting levels and covers the
allFieldsNotNullable bitmap-skip path. CompactBinaryRow.getStruct also
drops a per-call buffer.slice() allocation by pointing to the parent
buffer at the right offset instead.


- human_verification:  ✔️ 
- performance_verification: ✔️ 
- provenance_license_confirmation: ✔️ 

## Does this PR introduce any user-facing change?

Compact codec user enjoys more CPU and memory on business logic without
even realizing it.
No public API or wire changes.

## Benchmark
```
    RowFormatAllocationProbe bytes/op vs apache/main, compact mode,
    median of five:
      root    100208 -> 74305 (-25.9%)
      array     9032 ->  6629 (-26.6%)
      matrix   72005 -> 52760 (-26.7%)
      map       7744 ->  6401 (-17.3%)
      encode   55423 -> 41762 (-24.6%)
    Standard mode unchanged within noise.
```

---------

Co-authored-by: Claude (on behalf of Steven Schlansker) <stevenschlansker+claude@apache.org>

Author: stevenschlansker

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

@@ -29,7 +29,6 @@
 
 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;
@@ -38,12 +37,10 @@ class BinaryRowEncoder<T> implements RowEncoder<T> {
 
   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;
@@ -82,7 +79,7 @@ T decode(final MemoryBuffer buffer, final int size) {
               schema, schemaHash, peerSchemaHash));
     }
     final int rowSize = size - 8;
-    final BinaryRow row = codecFactory.newRow(schema);
+    final BinaryRow row = writer.newRow();
     row.pointTo(buffer, buffer.readerIndex(), rowSize);
     buffer.increaseReaderIndex(rowSize);
     return fromRow(row);

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

@@ -23,10 +23,8 @@
 import java.util.Map;
 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;
@@ -76,11 +74,6 @@ public MapEncoderBuilder newMapEncoder(
     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);

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

@@ -23,7 +23,6 @@
 import java.util.Map;
 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;
@@ -72,11 +71,6 @@ public MapEncoderBuilder newMapEncoder(
     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);

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

@@ -23,7 +23,6 @@
 import java.util.Map;
 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.type.Field;
@@ -47,8 +46,6 @@ ArrayEncoderBuilder newArrayEncoder(
 
   MapEncoderBuilder newMapEncoder(TypeRef<? extends Map<?, ?>> mapType, TypeRef<?> beanToken);
 
-  BinaryRow newRow(Schema schema);
-
   BinaryArray newArray(Field field);
 
   BinaryMap newMap(Field field);

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

@@ -63,7 +63,7 @@ Function<BaseBinaryRowWriter, RowEncoder<T>> buildForWriter() {
       @Override
       public RowEncoder<T> apply(final BaseBinaryRowWriter writer) {
         return new BinaryRowEncoder<T>(
-            schema, codecFormat, rowEncoderFactory.apply(writer), writer, sizeEmbedded);
+            schema, rowEncoderFactory.apply(writer), writer, sizeEmbedded);
       }
     };
   }

java/fory-format/src/main/java/org/apache/fory/format/row/binary/BinaryArray.java

@@ -58,12 +58,13 @@ public class BinaryArray extends UnsafeTrait implements ArrayData {
 
   public BinaryArray(Field field) {
     this(field, elementSize(field));
+    initializeExtData(1); // Only require at most one slot to cache the schema for array type.
   }
 
+  /** Skips {@code extData}; subclass must override {@code getStruct(int, Field, int)}. */
   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) {

java/fory-format/src/main/java/org/apache/fory/format/row/binary/BinaryRow.java

@@ -76,6 +76,14 @@ public BinaryRow(Schema schema) {
     initializeExtData(numFields);
   }
 
+  /** Skips {@code extData}; subclass must override {@code getStruct(int, Field, int)}. */
+  protected BinaryRow(Schema schema, int bitmapWidthInBytes) {
+    this.schema = schema;
+    this.numFields = schema.numFields();
+    Preconditions.checkArgument(numFields > 0);
+    this.bitmapWidthInBytes = bitmapWidthInBytes;
+  }
+
   protected int computeBitmapWidthInBytes() {
     return BitUtils.calculateBitmapWidthInBytes(numFields);
   }

java/fory-format/src/main/java/org/apache/fory/format/row/binary/CompactBinaryArray.java

@@ -24,7 +24,6 @@
 import org.apache.fory.format.row.binary.writer.CompactBinaryRowWriter;
 import org.apache.fory.format.type.DataTypes;
 import org.apache.fory.format.type.Field;
-import org.apache.fory.format.type.Schema;
 import org.apache.fory.memory.MemoryBuffer;
 
 public class CompactBinaryArray extends BinaryArray {
@@ -33,12 +32,30 @@ public class CompactBinaryArray extends BinaryArray {
   private final boolean elementNullable;
   private int headerInBytes;
 
+  /** {@code null} when the element is not a struct. */
+  private final CompactRowLayout elementLayout;
+
+  /** Cold path. Hot reads use {@link #CompactBinaryArray(Field, CompactRowLayout)}. */
   public CompactBinaryArray(final Field field) {
+    this(field, null);
+  }
+
+  /** {@code prebuiltElementLayout} may be {@code null}; built on demand for struct elements. */
+  CompactBinaryArray(final Field field, final CompactRowLayout prebuiltElementLayout) {
     super(field, CompactBinaryArrayWriter.elementWidth(field));
     DataTypes.ListType listType = (DataTypes.ListType) field.type();
     elementField = listType.valueField();
     fixedWidth = CompactBinaryRowWriter.fixedWidthFor(elementField);
     elementNullable = elementField.nullable();
+    if (elementField.type() instanceof DataTypes.StructType) {
+      elementLayout =
+          prebuiltElementLayout != null
+              ? prebuiltElementLayout
+              : new CompactRowLayout(
+                  CompactBinaryRowWriter.sortSchema(DataTypes.createSchema(elementField)));
+    } else {
+      elementLayout = null;
+    }
   }
 
   @Override
@@ -105,28 +122,18 @@ protected BinaryRow getStruct(final int ordinal, final Field field, final int ex
       return null;
     }
     assert field == elementField;
+    final CompactBinaryRow row = elementLayout.newRow();
     if (fixedWidth == -1) {
-      return super.getStruct(ordinal, field, extDataSlot);
-    }
-    if (extData[extDataSlot] == null) {
-      extData[extDataSlot] = newSchema(field);
+      final long offsetAndSize = getInt64(ordinal);
+      final int relativeOffset = (int) (offsetAndSize >> 32);
+      final int size = (int) offsetAndSize;
+      row.pointTo(getBuffer(), getBaseOffset() + relativeOffset, size);
+    } else {
+      row.pointTo(getBuffer(), getOffset(ordinal), fixedWidth);
     }
-    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);

java/fory-format/src/main/java/org/apache/fory/format/row/binary/CompactBinaryRow.java

@@ -19,8 +19,6 @@
 
 package org.apache.fory.format.row.binary;
 
-import org.apache.fory.format.row.binary.writer.CompactBinaryRowWriter;
-import org.apache.fory.format.type.DataTypes;
 import org.apache.fory.format.type.Field;
 import org.apache.fory.format.type.Schema;
 import org.apache.fory.memory.MemoryBuffer;
@@ -41,47 +39,39 @@
  * <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 CompactRowLayout layout;
   private final int bitmapOffset;
 
   public CompactBinaryRow(final Schema schema) {
-    this(schema, CompactBinaryRowWriter.fixedOffsets(schema));
+    this(new CompactRowLayout(schema));
   }
 
-  public CompactBinaryRow(final Schema schema, final int[] fixedOffsets) {
-    super(schema);
-    this.fixedOffsets = fixedOffsets;
-    bitmapOffset = fixedOffsets[fixedOffsets.length - 1];
-    allFieldsNotNullable = CompactBinaryRowWriter.allNotNullable(schema.fields());
+  CompactBinaryRow(CompactRowLayout layout) {
+    super(layout.schema, layout.bitmapWidthInBytes);
+    this.layout = layout;
+    this.bitmapOffset = layout.fixedOffsets[layout.fixedOffsets.length - 1];
   }
 
-  @Override
-  protected int computeBitmapWidthInBytes() {
-    // cannot use field due to initialization order
-    if (CompactBinaryRowWriter.allNotNullable(schema.fields())) {
-      return 0;
-    }
-    return CompactBinaryRowWriter.headerBytes(schema);
+  CompactRowLayout getLayout() {
+    return layout;
   }
 
   @Override
   public boolean isNullAt(final int ordinal) {
-    if (allFieldsNotNullable) {
+    if (layout.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];
+    return baseOffset + layout.fixedOffsets[ordinal];
   }
 
   @Override
   public MemoryBuffer getBuffer(final int ordinal) {
-    final int fixedWidthBinary = CompactBinaryRowWriter.fixedWidthFor(schema, ordinal);
+    final int fixedWidthBinary = layout.fixedWidths[ordinal];
     if (fixedWidthBinary >= 0) {
       if (isNullAt(ordinal)) {
         return null;
@@ -94,7 +84,7 @@ public MemoryBuffer getBuffer(final int ordinal) {
 
   @Override
   public byte[] getBinary(final int ordinal) {
-    final int fixedWidthBinary = CompactBinaryRowWriter.fixedWidthFor(schema, ordinal);
+    final int fixedWidthBinary = layout.fixedWidths[ordinal];
     if (fixedWidthBinary >= 0) {
       if (isNullAt(ordinal)) {
         return null;
@@ -112,27 +102,30 @@ protected BinaryRow getStruct(final int ordinal, final Field field, final int ex
     if (isNullAt(ordinal)) {
       return null;
     }
-    final int fixedWidthBinary = CompactBinaryRowWriter.fixedWidthFor(schema, ordinal);
+    final CompactBinaryRow row = layout.childLayouts[ordinal].newRow();
+    final int fixedWidthBinary = layout.fixedWidths[ordinal];
     if (fixedWidthBinary == -1) {
-      return super.getStruct(ordinal, field, extDataSlot);
-    }
-    if (extData[extDataSlot] == null) {
-      extData[extDataSlot] = DataTypes.createSchema(field);
+      final long offsetAndSize = getInt64(ordinal);
+      final int relativeOffset = (int) (offsetAndSize >> 32);
+      final int size = (int) offsetAndSize;
+      row.pointTo(getBuffer(), getBaseOffset() + relativeOffset, size);
+    } else {
+      row.pointTo(getBuffer(), getOffset(ordinal), fixedWidthBinary);
     }
-    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));
+  BinaryArray getArray(final int ordinal, final Field field) {
+    if (isNullAt(ordinal)) {
+      return null;
+    }
+    final long offsetAndSize = getInt64(ordinal);
+    final int relativeOffset = (int) (offsetAndSize >> 32);
+    final int size = (int) offsetAndSize;
+    final CompactBinaryArray array = new CompactBinaryArray(field, layout.childLayouts[ordinal]);
+    array.pointTo(getBuffer(), getBaseOffset() + relativeOffset, size);
+    return array;
   }
 
   @Override
@@ -152,6 +145,6 @@ protected int nullBitmapOffset() {
 
   @Override
   protected BinaryRow rowForCopy() {
-    return new CompactBinaryRow(schema, fixedOffsets);
+    return new CompactBinaryRow(layout);
   }
 }