perf(format): generate row-codec schema projections lazily on first decode

Schema-evolution projection codecs were compiled eagerly at builder time:
the row/array/map builders enumerated the full historical cross-product and
loaded one generated codec class per combination, so a deep nested version
history paid its whole class cost up front whether or not those versions ever
appeared on the wire.

Defer that compilation to decode. Each builder now builds an immutable
hash -> ProjectionSource index that holds only the inputs (the VersionedSchema
and codegen context); the codec class is compiled the first time a payload with
that hash is decoded and cached in a per-encoder LongMap. No new synchronization:
encoders are single-threaded, and the class compile is already memoized globally
by the shared code generator, so a concurrent first-miss on the same hash
compiles the class once.

The build-time collision guards stay eager over the full cross-product (row/array
via SchemaHistory's strict-hash guard, map via its two combined-hash guards), so a
hash clash still fails fast at build rather than on an unlucky decode.

Because the build-time class count no longer tracks annotated history, the
projection-count warning no longer flags a real cost; remove
PROJECTION_COUNT_WARN_THRESHOLD and warnIfManyProjections. Also fix stale Javadoc
that claimed map keys are always read at the current schema and that map support
is in progress, and update the row-format guide's generation-cost note to the
lazy model.

Author: Claude (on behalf of Steven Schlansker)

docs/guide/java/row-format.md

@@ -289,14 +289,13 @@ map may evolve more than one distinct bean class across its key and value. A top
 its own hash identifying both layouts together; a map nested inside an array, another map, or a
 bean field has its layouts folded into the enclosing payload's hash.
 
-When a versioned bean contains other versioned beans, the reader generates one projection codec
-class per combination of versions across the composition. The count grows as the product of the
-version counts of the distinct nested versioned bean classes, not the number of fields, so
-reusing a class across several fields adds no combinations. A map whose key and value both evolve
-multiplies their version counts the same way. If the product across distinct classes becomes a
-concern, drop entries from each bean's `History` interface once you no longer need to read payloads
-from that range. Retiring a history entry is purely a read-side decision; the writer always uses the
-current schema.
+When a versioned bean contains other versioned beans, the reader can read one projection layout per
+combination of versions across the composition. A reader compiles a combination's codec the first
+time it decodes a payload at that combination, so the cost tracks the historical versions you
+actually receive, not the number you could in principle define. A map whose key and value both
+evolve combines their versions the same way. Retiring an entry from a bean's `History` interface
+once you no longer read payloads from that range stops the reader from accepting those payloads; it
+is purely a read-side decision, and the writer always uses the current schema.
 
 ## Related Topics
 

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

@@ -23,7 +23,6 @@
 
 import java.lang.invoke.MethodHandle;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
@@ -104,62 +103,74 @@ private Function<BinaryArrayWriter, ArrayEncoder<C>> buildVersionedWithWriter(
     SchemaHistory history = buildElementSchemaHistory(elementField.name(), elementType);
     SchemaHistory.VersionedSchema current = history.current();
 
-    // Generate per-combination row codec classes and per-combination array codec classes. The
-    // suffix encodes each chosen inner-bean version so that distinct cross-product entries do not
-    // collide on a single generated class.
+    // Index of hash → deferred projection source per non-current combination. Building it compiles
+    // nothing: a combination's row and array codec classes are generated the first time a payload
+    // with that hash is decoded. The suffix encodes each chosen inner-bean version so distinct
+    // cross-product entries do not collide on a single generated class.
     //
     // Keyed by the raw strict hash straight from SchemaHistory, which already proves these hashes
     // are unique across versions() and distinct from the current schema, so no builder-side
     // collision check is needed here (unlike the map codec's combined (key, value) hash).
-    Map<Long, ProjectionArrayFactory> projectionFactories = new HashMap<>();
+    LongMap<BinaryArrayEncoder.ProjectionSource> projectionSources = new LongMap<>();
+    String elementName = elementField.name();
     for (SchemaHistory.VersionedSchema vs : history.versions()) {
       if (vs == current) {
         continue;
       }
-      String suffix = ProjectionRouting.projectionSuffix(vs);
-      // Generates the projection row codec for every nested versioned bean class in this
-      // combination, both map key and value, so the array codec's references all resolve.
-      Map<Class<?>, String> nestedSuffixes = ProjectionRouting.nestedSuffixesFor(vs, codecFormat);
-      Class<?> arrayClass =
-          Encoders.loadOrGenProjectionArrayCodecClass(
-              collectionType, TypeRef.of(elementClass), codecFormat, suffix, nestedSuffixes);
-      MethodHandle ctor = Encoders.constructorHandleFor(arrayClass, GeneratedArrayEncoder.class);
-      // forElement substitutes each chosen historical struct into its leaf, so the element field at
-      // this combination is simply the single field of vs.schema(); wrap it back in the list field.
-      Field histListField =
-          DataTypes.arrayField(elementField.name(), DataTypes.fieldOfSchema(vs.schema(), 0));
-      projectionFactories.put(vs.strictHash(), new ProjectionArrayFactory(histListField, ctor));
+      projectionSources.put(vs.strictHash(), new ProjectionSource(elementClass, elementName, vs));
     }
     final Function<BinaryArrayWriter, GeneratedArrayEncoder> currentFactory =
         generatedEncoderFactory();
     long currentHash = current.strictHash();
     return new Function<BinaryArrayWriter, ArrayEncoder<C>>() {
       @Override
       public ArrayEncoder<C> apply(final BinaryArrayWriter writer) {
-        LongMap<BinaryArrayEncoder.ProjectionArrayCodec> proj =
-            new LongMap<>(projectionFactories.size());
-        for (Map.Entry<Long, ProjectionArrayFactory> entry : projectionFactories.entrySet()) {
-          proj.put(entry.getKey(), entry.getValue().instantiate(fory));
-        }
         return new BinaryArrayEncoder<>(
-            writer, currentFactory.apply(writer), sizeEmbedded, currentHash, proj);
+            writer,
+            currentFactory.apply(writer),
+            sizeEmbedded,
+            currentHash,
+            projectionSources,
+            fory);
       }
     };
   }
 
-  private final class ProjectionArrayFactory {
-    private final Field elementField;
-    private final MethodHandle ctor;
-
-    ProjectionArrayFactory(Field elementField, MethodHandle ctor) {
-      this.elementField = elementField;
-      this.ctor = ctor;
+  /**
+   * Deferred projection codec for one historical element version. Holds only the inputs to generate
+   * the codec; the row and array codec classes are compiled on the first {@link #compile} call (the
+   * first decode of this version's hash), not at build time.
+   */
+  private final class ProjectionSource implements BinaryArrayEncoder.ProjectionSource {
+    private final Class<?> elementClass;
+    private final String elementName;
+    private final SchemaHistory.VersionedSchema version;
+
+    ProjectionSource(
+        Class<?> elementClass, String elementName, SchemaHistory.VersionedSchema version) {
+      this.elementClass = elementClass;
+      this.elementName = elementName;
+      this.version = version;
     }
 
-    BinaryArrayEncoder.ProjectionArrayCodec instantiate(Fory fory) {
+    @Override
+    public BinaryArrayEncoder.ProjectionArrayCodec compile(Fory fory) {
+      String suffix = ProjectionRouting.projectionSuffix(version);
+      // Generates the projection row codec for every nested versioned bean class in this
+      // combination, both map key and value, so the array codec's references all resolve.
+      Map<Class<?>, String> nestedSuffixes =
+          ProjectionRouting.nestedSuffixesFor(version, codecFormat);
+      Class<?> arrayClass =
+          Encoders.loadOrGenProjectionArrayCodecClass(
+              collectionType, TypeRef.of(elementClass), codecFormat, suffix, nestedSuffixes);
+      MethodHandle ctor = Encoders.constructorHandleFor(arrayClass, GeneratedArrayEncoder.class);
+      // forElement substitutes each chosen historical struct into its leaf, so the element field at
+      // this combination is simply the single field of vs.schema(); wrap it back in the list field.
+      Field histListField =
+          DataTypes.arrayField(elementName, DataTypes.fieldOfSchema(version.schema(), 0));
       try {
-        BinaryArrayWriter projWriter = codecFormat.newArrayWriter(elementField);
-        Object[] references = {elementField, projWriter, fory};
+        BinaryArrayWriter projWriter = codecFormat.newArrayWriter(histListField);
+        Object[] references = {histListField, projWriter, fory};
         GeneratedArrayEncoder codec = (GeneratedArrayEncoder) ctor.invokeExact(references);
         return new BinaryArrayEncoder.ProjectionArrayCodec(projWriter, codec);
       } catch (Throwable e) {

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

@@ -26,22 +26,10 @@
 import org.apache.fory.format.type.CustomTypeEncoderRegistry;
 import org.apache.fory.format.type.Schema;
 import org.apache.fory.format.type.SchemaHistory;
-import org.apache.fory.logging.Logger;
-import org.apache.fory.logging.LoggerFactory;
 import org.apache.fory.reflect.TypeRef;
 import org.apache.fory.type.TypeResolutionContext;
 
 public class BaseCodecBuilder<B extends BaseCodecBuilder<B>> {
-  private static final Logger LOG = LoggerFactory.getLogger(BaseCodecBuilder.class);
-
-  /**
-   * Number of historical schemas for one bean above which {@link #buildSchemaHistory} logs a
-   * warning. Each distinct schema becomes one generated projection codec class (compiled and loaded
-   * at build time), and the count grows as the product of the per-class version counts across
-   * nested versioned beans. The JVM handles far more classes than this; the threshold flags a
-   * likely misconfigured version history, since no hand-written history reaches it by accident.
-   */
-  private static final int PROJECTION_COUNT_WARN_THRESHOLD = 256;
 
   protected Schema schema;
   protected int initialBufferSize = 16;
@@ -115,9 +103,7 @@ public B compactEncoding() {
    * unchanged.
    */
   protected SchemaHistory buildSchemaHistory(final Class<?> targetClass) {
-    SchemaHistory history = SchemaHistory.build(targetClass, schemaTransform());
-    warnIfManyProjections(targetClass.getName(), history);
-    return history;
+    return SchemaHistory.build(targetClass, schemaTransform());
   }
 
   /**
@@ -129,19 +115,6 @@ protected SchemaHistory buildSchemaHistory(final Class<?> targetClass) {
    */
   protected SchemaHistory buildElementSchemaHistory(
       final String fieldName, final TypeRef<?> elementType) {
-    SchemaHistory history = elementSchemaHistory(fieldName, elementType);
-    warnIfManyProjections("element " + elementType, history);
-    return history;
-  }
-
-  /**
-   * Builds a position's history without the per-position projection-count warning. The map path
-   * uses this and warns once on the key/value cross-product, which is the count of classes it
-   * generates; the array path uses {@link #buildElementSchemaHistory}, whose per-position count is
-   * exact.
-   */
-  protected SchemaHistory elementSchemaHistory(
-      final String fieldName, final TypeRef<?> elementType) {
     return SchemaHistory.forElement(fieldName, elementType, schemaTransform());
   }
 
@@ -161,27 +134,6 @@ private UnaryOperator<Schema> schemaTransform() {
         : UnaryOperator.identity();
   }
 
-  private static void warnIfManyProjections(final String label, final SchemaHistory history) {
-    warnIfManyProjections(label, history.versions().size());
-  }
-
-  /**
-   * Warn when {@code projectionCount} generated projection codec classes exceeds the threshold. The
-   * map path passes the key/value cross-product here, not a single position's count, because that
-   * product is the number of classes it actually generates.
-   */
-  protected static void warnIfManyProjections(final String label, final int projectionCount) {
-    if (projectionCount > PROJECTION_COUNT_WARN_THRESHOLD) {
-      LOG.warn(
-          "Schema evolution for {} resolved {} historical schemas, each generating a projection "
-              + "codec class. This count grows as the product of per-class version counts across "
-              + "nested versioned beans; retire @ForyVersion history ranges you no longer read to "
-              + "reduce it.",
-          label,
-          projectionCount);
-    }
-  }
-
   @SuppressWarnings("unchecked")
   protected B castThis() {
     return (B) this;

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

@@ -19,6 +19,7 @@
 
 package org.apache.fory.format.encoder;
 
+import org.apache.fory.Fory;
 import org.apache.fory.collection.LongMap;
 import org.apache.fory.exception.ClassNotCompatibleException;
 import org.apache.fory.format.row.binary.BinaryArray;
@@ -39,9 +40,23 @@ class BinaryArrayEncoder<T> implements ArrayEncoder<T> {
    */
   private final long currentHash;
 
-  /** Per-version projection codecs and their element fields. {@code null} disables versioning. */
+  /**
+   * Hash → source able to compile a projection codec for an older element schema. {@code null}
+   * disables versioning (evolution off, no hash prefix). Non-null but empty under evolution with no
+   * historical versions: the hash prefix is still written for flag-mismatch detection. Shared and
+   * immutable; a combination's codec class is compiled only the first time its hash is decoded.
+   */
+  private final LongMap<ProjectionSource> projectionSources;
+
+  /**
+   * Per-encoder cache of projection codecs compiled on first decode of their hash. Lock-free: an
+   * encoder is single-threaded (see {@link ArrayEncoder}), and the class compile is memoized
+   * globally by the shared code generator.
+   */
   private final LongMap<ProjectionArrayCodec> projections;
 
+  private final Fory fory;
+
   /**
    * A projection variant of the array codec along with the writer used to materialize an array
    * instance of the right physical type (standard vs. compact) for the historical element field.
@@ -56,24 +71,32 @@ static final class ProjectionArrayCodec {
     }
   }
 
+  /** Compiles one historical element version's projection codec on first decode of its hash. */
+  interface ProjectionSource {
+    ProjectionArrayCodec compile(Fory fory);
+  }
+
   BinaryArrayEncoder(
       final BinaryArrayWriter writer,
       final GeneratedArrayEncoder codec,
       final boolean sizeEmbedded) {
-    this(writer, codec, sizeEmbedded, 0L, null);
+    this(writer, codec, sizeEmbedded, 0L, null, null);
   }
 
   BinaryArrayEncoder(
       final BinaryArrayWriter writer,
       final GeneratedArrayEncoder codec,
       final boolean sizeEmbedded,
       final long currentHash,
-      final LongMap<ProjectionArrayCodec> projections) {
+      final LongMap<ProjectionSource> projectionSources,
+      final Fory fory) {
     this.writer = writer;
     this.codec = codec;
     this.sizeEmbedded = sizeEmbedded;
     this.currentHash = currentHash;
-    this.projections = projections;
+    this.projectionSources = projectionSources;
+    this.fory = fory;
+    this.projections = projectionSources == null ? null : new LongMap<>(projectionSources.size);
   }
 
   @Override
@@ -107,7 +130,7 @@ public T decode(final byte[] bytes) {
 
   @SuppressWarnings("unchecked")
   T decode(final MemoryBuffer buffer, final int size) {
-    if (projections == null) {
+    if (projectionSources == null) {
       // Evolution off: the whole payload is body, with no hash prefix. Reading evolution-on bytes
       // here cannot be caught: the array wire form has no hash slot when evolution is off, and an
       // evolution-on payload's leading 8-byte FNV hash is indistinguishable from a valid array
@@ -132,7 +155,7 @@ T decode(final MemoryBuffer buffer, final int size) {
       buffer.readerIndex(readerIndex + bodySize);
       return fromArray(array);
     }
-    ProjectionArrayCodec projection = projections.get(peerHash);
+    ProjectionArrayCodec projection = resolveProjection(peerHash);
     if (projection == null) {
       throw new ClassNotCompatibleException(
           String.format(
@@ -146,10 +169,29 @@ T decode(final MemoryBuffer buffer, final int size) {
     return (T) projection.codec.fromArray(array);
   }
 
+  /**
+   * The projection codec for {@code peerHash}, or {@code null} if no historical version has that
+   * hash. Compiles the codec on first encounter and caches it. Single-threaded by the {@link
+   * ArrayEncoder} contract, so the cache needs no locking.
+   */
+  private ProjectionArrayCodec resolveProjection(final long peerHash) {
+    ProjectionArrayCodec cached = projections.get(peerHash);
+    if (cached != null) {
+      return cached;
+    }
+    ProjectionSource source = projectionSources.get(peerHash);
+    if (source == null) {
+      return null;
+    }
+    ProjectionArrayCodec projection = source.compile(fory);
+    projections.put(peerHash, projection);
+    return projection;
+  }
+
   @Override
   public byte[] encode(final T obj) {
     final BinaryArray array = toArray(obj);
-    if (projections == null) {
+    if (projectionSources == null) {
       return writer.getBuffer().getBytes(0, array.getSizeInBytes());
     }
     // Build the result with a single allocation: the result byte[]. The hash header is poked
@@ -172,7 +214,7 @@ public int encode(final MemoryBuffer buffer, final T obj) {
     if (sizeEmbedded) {
       buffer.writeInt32(-1);
     }
-    if (projections != null) {
+    if (projectionSources != null) {
       buffer.writeInt64(currentHash);
     }
     try {