Draft: feat(format): schema evolution for the Java row codec

docs/guide/java/row-format.md

@@ -113,6 +113,78 @@ Row format is ideal for:
 - **Data pipelines**: Processing data without full object reconstruction
 - **Cross-language data sharing**: When data needs to be accessed from multiple languages
 
+## Schema evolution
+
+Enable `.withSchemaEvolution()` on a row, array, or map codec builder to read payloads written
+by older versions of the same bean. Writing always uses the current version; reading detects
+the payload's version from a strict hash at the head of the payload. Java only.
+
+Annotate fields added after v1 with `@ForyVersion(since = N)`:
+
+```java
+@Data
+public class Person {
+  private String name;
+  private int age;
+
+  @ForyVersion(since = 2)
+  private String email;
+}
+```
+
+A v1 payload (with `name` and `age` only) decodes to a `Person` whose `email` is `null`.
+Primitive fields added later default to `0`, `0.0`, or `false`. If a class adopts versioning
+after its v1 is already in the wild, set `@ForySchema(baseVersion = N)` so unannotated fields
+are treated as present since version `N`.
+
+Remove a field by deleting the Java member and declaring it on a nested history interface as a
+method with a `@ForyVersion(until = N)`. The method's return type carries any parameterized
+type information from the original field.
+
+```java
+@Data
+@ForySchema(removedFields = Person.History.class)
+public class Person {
+  private String name;
+
+  @ForyVersion(since = 2)
+  private String email;
+
+  interface History {
+    @ForyVersion(until = 3)
+    int age();
+
+    @ForyVersion(until = 5)
+    List<String> tags();
+  }
+}
+```
+
+The history method name matches the original live descriptor name: the field name for Lombok
+`@Data` or records (`age`, `tags`), or the full accessor name for JavaBeans-style classes and
+interfaces (`getAge`).
+
+### Wire format and limitations
+
+Producers and consumers must agree on the `withSchemaEvolution()` flag — they are not
+wire-compatible otherwise. Row payloads always carry an 8-byte hash slot; under evolution its
+value is the strict hash (which includes field name and nullability), so a flag-mismatched
+peer fails loudly with `ClassNotCompatibleException`. Arrays and maps of bean elements prepend
+an 8-byte strict-hash prefix under evolution and no prefix otherwise; an evolution-on consumer
+reading evolution-off bytes also fails with `ClassNotCompatibleException`, but the reverse
+direction (evolution-off consumer, evolution-on bytes) is undefined.
+
+Cross-language consumers (Python, C++) cannot read evolution-enabled payloads.
+
+Map keys do not carry a per-payload hash; a versioned bean used as a map key is read with the
+current schema only, not dispatched to a projection codec.
+
+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
+per-bean version counts. If that 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.
+
 ## Cross-Language Compatibility
 
 Row format works seamlessly across languages. The same binary data can be accessed from:

java/fory-format/src/main/java/org/apache/fory/format/annotation/ForySchema.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.annotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Class-level row-codec schema metadata used when the codec builder enables schema evolution.
+ *
+ * <p>{@link #baseVersion()} sets the implicit {@code since} for live fields that lack a
+ * {@link ForyVersion} annotation; this lets a class adopt versioning mid-history without having
+ * to annotate every existing member.
+ *
+ * <p>{@link #removedFields()} points at a class (conventionally a nested {@code interface}) whose
+ * accessor methods describe fields that have been removed from this bean but still appear on the
+ * wire in older payloads. Each method's return type is the original Java type of the removed
+ * field; each method must carry a {@link ForyVersion} annotation with {@code until} set, since
+ * removed fields have a known end-of-life version.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * @Data
+ * @ForySchema(removedFields = MyBean.History.class)
+ * public class MyBean {
+ *   private String name;
+ *
+ *   interface History {
+ *     @ForyVersion(until = 3)
+ *     List<String> tags();
+ *
+ *     @ForyVersion(since = 2, until = 5)
+ *     Map<String, Long> counters();
+ *   }
+ * }
+ * }</pre>
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+public @interface ForySchema {
+  int baseVersion() default 1;
+
+  /**
+   * A class whose accessor methods describe historically-present-but-now-removed fields. Default
+   * {@code void.class} means there are no removed fields. The class is never instantiated; the
+   * codec reads its method signatures and annotations.
+   */
+  Class<?> removedFields() default void.class;
+}

java/fory-format/src/main/java/org/apache/fory/format/annotation/ForyVersion.java

@@ -0,0 +1,44 @@
+/*
+ * 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.annotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Declares the version window in which a row-codec field is logically present. The window is
+ * inclusive on the left and exclusive on the right, so {@code since=2, until=5} means versions 2,
+ * 3, and 4.
+ *
+ * <p>Only effective when the codec builder is configured with
+ * {@code withSchemaEvolution()}; otherwise the annotation is ignored and the field is treated as
+ * always present.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.FIELD, ElementType.METHOD, ElementType.RECORD_COMPONENT})
+public @interface ForyVersion {
+  /** First version (inclusive) that contains this field. Defaults to the class base version. */
+  int since() default 1;
+
+  /** First version (exclusive) that no longer contains this field. */
+  int until() default Integer.MAX_VALUE;
+}

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

@@ -22,18 +22,25 @@
 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.HashMap;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Set;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.function.UnaryOperator;
+import org.apache.fory.Fory;
 import org.apache.fory.format.row.binary.writer.BinaryArrayWriter;
+import org.apache.fory.format.row.binary.writer.CompactBinaryRowWriter;
+import org.apache.fory.format.type.CustomTypeEncoderRegistry;
 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.format.type.SchemaHistory;
 import org.apache.fory.format.type.TypeInference;
 import org.apache.fory.reflect.TypeRef;
+import org.apache.fory.type.TypeResolutionContext;
 import org.apache.fory.type.TypeUtils;
 import org.apache.fory.util.ExceptionUtils;
 
@@ -63,17 +70,108 @@ public ArrayEncoder<C> get() {
 
   Function<BinaryArrayWriter, ArrayEncoder<C>> buildWithWriter() {
     loadArrayInnerCodecs();
-    final Function<BinaryArrayWriter, GeneratedArrayEncoder> generatedEncoderFactory =
+    if (!schemaEvolution || !isBeanElement()) {
+      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);
+        }
+      };
+    }
+    return buildVersionedWithWriter();
+  }
+
+  /**
+   * True if the element is a bean — the only case where schema evolution affects the wire
+   * format. Unversioned beans still take the evolution path so the strict-hash prefix is always
+   * present and an evolution-on consumer can detect a flag-mismatched producer cleanly.
+   */
+  private boolean isBeanElement() {
+    Class<?> elementClass = getRawType(TypeUtils.getElementType(collectionType));
+    // Use the same resolution context as the row-format type inference, which synthesizes
+    // interface-typed bean fields. Without this, classes that contain interface members
+    // would not be recognized as beans even though the row codec can encode them.
+    return TypeUtils.isBean(
+        TypeRef.of(elementClass),
+        new TypeResolutionContext(CustomTypeEncoderRegistry.customTypeHandler(), true));
+  }
+
+  private Function<BinaryArrayWriter, ArrayEncoder<C>> buildVersionedWithWriter() {
+    Class<?> elementClass = getRawType(TypeUtils.getElementType(collectionType));
+    UnaryOperator<Schema> schemaTransform =
+        codecFormat == CompactCodecFormat.INSTANCE
+            ? CompactBinaryRowWriter::sortSchema
+            : UnaryOperator.identity();
+    SchemaHistory history = SchemaHistory.build(elementClass, schemaTransform);
+    SchemaHistory.VersionedSchema current = history.current();
+
+    // Make sure the current-version row codec class is generated.
+    Encoders.loadOrGenRowCodecClass(elementClass, codecFormat);
+    // Generate per-combination row codec classes and per-combination array codec classes. The
+    // suffix encodes the outer version plus each chosen inner-bean version so that distinct
+    // cross-product entries do not collide on a single generated class.
+    Map<Long, ProjectionArrayFactory> projectionFactories = new HashMap<>();
+    for (SchemaHistory.VersionedSchema vs : history.versions()) {
+      if (vs == current) {
+        continue;
+      }
+      String suffix = ProjectionRouting.projectionSuffix(vs);
+      Map<Class<?>, String> nestedSuffixes = ProjectionRouting.nestedSuffixesFor(vs, codecFormat);
+      Encoders.loadOrGenProjectionRowCodecClass(
+          elementClass, codecFormat, vs.schema(), vs.liveFieldNames(), suffix, nestedSuffixes);
+      Class<?> arrayClass =
+          Encoders.loadOrGenProjectionArrayCodecClass(
+              collectionType, TypeRef.of(elementClass), codecFormat, suffix);
+      MethodHandle ctor = Encoders.constructorHandleFor(arrayClass, GeneratedArrayEncoder.class);
+      // The array's "elementField" is a ListType whose valueField is the element struct. Build
+      // a parallel ListType for this historical version so the projection codec can produce a
+      // BinaryArray with the right element width.
+      Field histValueField =
+          DataTypes.field(
+              DataTypes.ARRAY_ITEM_NAME, new DataTypes.StructType(vs.schema().fields()), true);
+      Field histListField = DataTypes.arrayField(elementField.name(), histValueField);
+      projectionFactories.put(vs.strictHash(), new ProjectionArrayFactory(histListField, ctor));
+    }
+    final Function<BinaryArrayWriter, GeneratedArrayEncoder> currentFactory =
         generatedEncoderFactory();
+    long currentHash = current.strictHash();
     return new Function<BinaryArrayWriter, ArrayEncoder<C>>() {
       @Override
       public ArrayEncoder<C> apply(final BinaryArrayWriter writer) {
+        Map<Long, BinaryArrayEncoder.ProjectionArrayCodec> proj = new HashMap<>();
+        for (Map.Entry<Long, ProjectionArrayFactory> entry : projectionFactories.entrySet()) {
+          proj.put(entry.getKey(), entry.getValue().instantiate(fory));
+        }
         return new BinaryArrayEncoder<>(
-            writer, generatedEncoderFactory.apply(writer), sizeEmbedded);
+            writer, currentFactory.apply(writer), sizeEmbedded, currentHash, proj);
       }
     };
   }
 
+  private final class ProjectionArrayFactory {
+    private final Field elementField;
+    private final MethodHandle ctor;
+
+    ProjectionArrayFactory(Field elementField, MethodHandle ctor) {
+      this.elementField = elementField;
+      this.ctor = ctor;
+    }
+
+    BinaryArrayEncoder.ProjectionArrayCodec instantiate(Fory fory) {
+      try {
+        BinaryArrayWriter projWriter = codecFormat.newArrayWriter(elementField);
+        Object[] references = {elementField, projWriter, fory};
+        GeneratedArrayEncoder codec = (GeneratedArrayEncoder) ctor.invokeExact(references);
+        return new BinaryArrayEncoder.ProjectionArrayCodec(projWriter, codec);
+      } catch (Throwable e) {
+        throw ExceptionUtils.throwException(e);
+      }
+    }
+  }
+
   private void loadArrayInnerCodecs() {
     final Set<TypeRef<?>> set = new HashSet<>();
     Encoders.findBeanToken(collectionType, set);
@@ -90,30 +188,15 @@ 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);
-    }
+    final MethodHandle constructorHandle =
+        Encoders.constructorHandleFor(arrayCodecClass, GeneratedArrayEncoder.class);
     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) {
+        } catch (Throwable t) {
           throw ExceptionUtils.throwException(t);
         }
       }

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

@@ -54,7 +54,17 @@ public ArrayEncoderBuilder(Class<?> arrayCls, Class<?> beanClass) {
   }
 
   public ArrayEncoderBuilder(TypeRef<?> clsType, TypeRef<?> beanType) {
+    this(clsType, beanType, null);
+  }
+
+  /**
+   * Construct an array codec builder that embeds row codec class references for its element bean
+   * with the supplied suffix. Used by schema-evolution code to point per-version array codecs at
+   * per-version row codecs.
+   */
+  ArrayEncoderBuilder(TypeRef<?> clsType, TypeRef<?> beanType, String rowCodecSuffix) {
     super(new CodegenContext(), beanType);
+    this.rowCodecSuffixForBeans = rowCodecSuffix;
     arrayToken = clsType;
     ctx.reserveName(ROOT_ARRAY_WRITER_NAME);
     ctx.reserveName(ROOT_ARRAY_NAME);
@@ -83,7 +93,9 @@ public ArrayEncoderBuilder(TypeRef<?> clsType, TypeRef<?> beanType) {
   @Override
   public String genCode() {
     ctx.setPackage(CodeGenerator.getPackage(beanClass));
-    String className = codecClassName(beanClass, TypeInference.inferTypeName(arrayToken));
+    String className =
+        codecClassName(beanClass, TypeInference.inferTypeName(arrayToken))
+            + (rowCodecSuffixForBeans == null ? "" : rowCodecSuffixForBeans);
     ctx.setClassName(className);
     // don't addImport(arrayClass), because user class may name collide.
     // janino don't support generics, so GeneratedCodec has no generics