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

benchmarks/java/src/main/java/org/apache/fory/benchmark/SchemaEvolutionSuite.java

@@ -0,0 +1,162 @@
+/*
+ * 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.benchmark;
+
+import java.util.Arrays;
+import org.apache.fory.format.annotation.ForyVersion;
+import org.apache.fory.format.encoder.Encoders;
+import org.apache.fory.format.encoder.RowEncoder;
+import org.apache.fory.logging.Logger;
+import org.apache.fory.logging.LoggerFactory;
+import org.openjdk.jmh.Main;
+import org.openjdk.jmh.annotations.Benchmark;
+
+/**
+ * Row-codec schema-evolution throughput and allocation. Pair with the JMH gc profiler ({@code -prof
+ * gc}) to read {@code gc.alloc.rate.norm} (bytes per op). Two comparisons matter: {@code
+ * currentDecode} vs {@code olderDecode} shows that decoding an older payload through a projection
+ * codec allocates no more than decoding the current schema, because each projection holds its
+ * historical schema's row layout (no per-decode rebuild); and the {@code *NoEvolution} benchmarks
+ * vs their evolution-on counterparts show the steady-state cost of enabling {@code
+ * withSchemaEvolution()} when reading and writing current-version data.
+ */
+public class SchemaEvolutionSuite {
+  private static final Logger LOG = LoggerFactory.getLogger(SchemaEvolutionSuite.class);
+
+  public static class PersonV1 {
+    String name;
+    int age;
+  }
+
+  public static class PersonV2 {
+    String name;
+    int age;
+
+    @ForyVersion(since = 2)
+    String email;
+  }
+
+  // Evolution-enabled codecs for the current (V2) schema; the V1 codec only produces a payload
+  // whose hash routes the V2 reader onto its projection path. Both standard and compact formats
+  // are measured: compact is where a per-projection cached row layout matters, so olderDecode vs
+  // currentDecode there is the parity check.
+  private static final RowEncoder<PersonV1> v1Codec =
+      Encoders.buildBeanCodec(PersonV1.class).withSchemaEvolution().build().get();
+  private static final RowEncoder<PersonV2> v2Codec =
+      Encoders.buildBeanCodec(PersonV2.class).withSchemaEvolution().build().get();
+  private static final RowEncoder<PersonV1> v1CompactCodec =
+      Encoders.buildBeanCodec(PersonV1.class).compactEncoding().withSchemaEvolution().build().get();
+  private static final RowEncoder<PersonV2> v2CompactCodec =
+      Encoders.buildBeanCodec(PersonV2.class).compactEncoding().withSchemaEvolution().build().get();
+
+  // Evolution-disabled codecs for the same current (V2) schema. Comparing the *NoEvolution
+  // benchmarks against their evolution-on counterparts isolates the steady-state cost of the
+  // withSchemaEvolution() flag on the common path (reading and writing current-version data): the
+  // 8-byte hash slot the evolution wire format adds, plus the hash compare on decode.
+  private static final RowEncoder<PersonV2> v2PlainCodec =
+      Encoders.buildBeanCodec(PersonV2.class).build().get();
+  private static final RowEncoder<PersonV2> v2PlainCompactCodec =
+      Encoders.buildBeanCodec(PersonV2.class).compactEncoding().build().get();
+
+  private static final PersonV2 person = newPerson();
+  private static final byte[] currentBytes = v2Codec.encode(person);
+  private static final byte[] olderBytes = v1Codec.encode(newPersonV1());
+  private static final byte[] currentCompactBytes = v2CompactCodec.encode(person);
+  private static final byte[] olderCompactBytes = v1CompactCodec.encode(newPersonV1());
+  private static final byte[] plainBytes = v2PlainCodec.encode(person);
+  private static final byte[] plainCompactBytes = v2PlainCompactCodec.encode(person);
+
+  private static PersonV2 newPerson() {
+    PersonV2 p = new PersonV2();
+    p.name = "Ada Lovelace";
+    p.age = 36;
+    p.email = "ada@example.com";
+    return p;
+  }
+
+  private static PersonV1 newPersonV1() {
+    PersonV1 p = new PersonV1();
+    p.name = "Ada Lovelace";
+    p.age = 36;
+    return p;
+  }
+
+  @Benchmark
+  public Object encode() {
+    return v2Codec.encode(person);
+  }
+
+  @Benchmark
+  public Object currentDecode() {
+    return v2Codec.decode(currentBytes);
+  }
+
+  @Benchmark
+  public Object olderDecode() {
+    return v2Codec.decode(olderBytes);
+  }
+
+  @Benchmark
+  public Object compactEncode() {
+    return v2CompactCodec.encode(person);
+  }
+
+  @Benchmark
+  public Object compactCurrentDecode() {
+    return v2CompactCodec.decode(currentCompactBytes);
+  }
+
+  @Benchmark
+  public Object compactOlderDecode() {
+    return v2CompactCodec.decode(olderCompactBytes);
+  }
+
+  // Evolution-off baselines for the current path. Pair each with its evolution-on counterpart
+  // (encode/currentDecode and the compact variants) to read the flag's overhead.
+  @Benchmark
+  public Object encodeNoEvolution() {
+    return v2PlainCodec.encode(person);
+  }
+
+  @Benchmark
+  public Object currentDecodeNoEvolution() {
+    return v2PlainCodec.decode(plainBytes);
+  }
+
+  @Benchmark
+  public Object compactEncodeNoEvolution() {
+    return v2PlainCompactCodec.encode(person);
+  }
+
+  @Benchmark
+  public Object compactCurrentDecodeNoEvolution() {
+    return v2PlainCompactCodec.decode(plainCompactBytes);
+  }
+
+  public static void main(String[] args) throws Exception {
+    if (args.length == 0) {
+      String commandLine =
+          "org.apache.fory.*SchemaEvolutionSuite.* -f 3 -wi 3 -i 3 -t 1 -w 2s -r 2s -prof gc -rf csv";
+      args = commandLine.split(" ");
+    }
+    LOG.info("command line: {}", Arrays.toString(args));
+    Main.main(args);
+  }
+}

docs/row-format/java.md

@@ -277,6 +277,141 @@ BinaryRow row = encoder.toRow(child);
 Parent decoded = encoder.fromRow(row);
 ```
 
+## 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 {
+  String name;
+  int age;
+
+  @ForyVersion(since = 2)
+  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`. Unannotated fields are treated
+as present from the first version, so a class can adopt versioning by annotating only the fields
+added after v1.
+
+For a record, the absent component's default is passed to the canonical constructor, so a
+constructor that rejects `null` for a reference component added in a later version throws when
+decoding an older payload. Let the constructor tolerate the missing value, for example by
+normalizing `null` to a default:
+
+```java
+public record Person(String name, @ForyVersion(since = 2) String email) {
+  public Person {
+    if (email == null) {
+      email = "";
+    }
+  }
+}
+```
+
+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 {
+  String name;
+
+  @ForyVersion(since = 2)
+  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. For field-backed beans
+(Lombok `@Data`, records, or plain classes with a backing field) that is the field name
+(`age`, `tags`). For interface beans, where the live member is a getter with no backing field,
+it is the method name (`getAge`).
+
+A field handled by a custom codec evolves as its encoded representation. A codec that encodes
+to another type descends that type, so a versioned bean reachable only through the codec is
+still enumerated; a codec that supplies its own terminal `foryField` column replaces the whole
+field, so the declared type is never introspected and its members are invisible to the
+evolution walk.
+
+### 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.
+
+To adopt the flag on an existing deployment, enable `withSchemaEvolution()` on both sides in a
+release that changes no schema, then start evolving schemas only once every peer is on the
+evolution-enabled build. Turning the flag on and changing a schema in the same release strands
+any peer that has not yet upgraded.
+
+Cross-language consumers (Python, C++) cannot read evolution-enabled payloads today. The hash
+algorithm, wire names, field order, and framing are specified language-neutrally in the
+[row format specification](../specification/row_format_spec.md#schema-evolution) so other runtimes
+can implement them.
+
+Field identity on the wire is the snake_case form of the member name (`createdAt` becomes
+`created_at`), and a versioned layout orders fields by that wire name. For ordinary member names
+this is the same order the generated codecs already use. A member name with a literal underscore
+next to a case boundary makes the two orders diverge (`aB` sorts before `a_a` as a member name,
+but its wire name `a_b` sorts after `a_a`); such a bean is rejected when the codec is built, with
+a message naming the members to rename.
+
+Because field names are wire identity, a rename is a removal plus an addition, not an in-place
+mapping: declare the old name on the history interface with `until = N` and annotate the renamed
+member with `since = N`. Older payloads then decode with the renamed field at its default, not
+carrying over the old field's value; renaming without those declarations changes the payload hash
+and older payloads stop decoding. The member the wire name derives from is whatever backs the
+descriptor — the field for field-backed beans, the accessor method for interface beans — so
+converting a bean between the two styles also changes every wire name and is a rename of every
+field.
+
+A reader selects the matching layout from the 8-byte strict hash on the payload. The hash includes
+field names and nullability and is checked for collisions across a bean's own versions when the
+codec is built, but it is still a 64-bit value: a payload whose hash coincides with one of the
+reader's historical layouts is decoded against that layout. This is the same hash-based dispatch
+the row format has always used, so feeding a codec bytes it was not built for has undefined results
+whether or not evolution is enabled. Only hand a codec payloads produced for the same bean.
+
+Nested evolution works to arbitrary depth and places no restriction on shape: a versioned bean
+may contain versioned beans that themselves contain versioned beans, the same versioned bean
+class may back more than one field, and fields typed as a non-evolving bean, a list, or a map are
+unrestricted. Each nesting level is routed to the correct historical layout. A versioned bean may
+be used as a map key as well as a map value, and the key and value evolve independently. This
+holds wherever the map appears: as the codec's top-level type, nested inside a bean field, or
+reached through a top-level array or map (such as `List<Map<KeyBean, ValueBean>>`), and a single
+map may evolve more than one distinct bean class across its key and value. A top-level map carries
+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 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
 
 - [Cross-Language Interoperability](../object-serialization/java/basic-serialization.md#cross-language-interoperability) - xlang mode

docs/specification/row_format_spec.md

@@ -345,6 +345,94 @@ if (fixed_width % 8 == 0):
 
 ---
 
+## Schema Evolution
+
+Schema evolution lets a codec read payloads written by older versions of the same struct. It is
+currently implemented in Java only and does not change the cross-language wire contract above;
+producer and consumer must agree on whether it is enabled. The protocol below is defined over
+language-neutral inputs so another runtime can implement it and interoperate with Java payloads.
+
+See the [Java row format guide](../row-format/java.md#schema-evolution) for usage, annotations,
+and limitations.
+
+### Version model
+
+- Versions are per-struct integers starting at 1. Version numbers never appear on the wire;
+  payloads are identified by a strict schema hash, so peers need identical declared histories,
+  not synchronized version counters.
+- A schema history is a set of field entries `(wire name, type, nullability, since, until)`.
+  A live field has an unbounded `until`; a removed field declares the version it disappeared in.
+  How the entries are declared is language-specific (Java uses annotations and a removed-fields
+  interface); the protocol object is the entry tuple.
+- A layout is materialized at every version where the active field set changes; layouts that come
+  out identical collapse into one entry.
+- Reading a payload that predates a field yields null for a nullable field and the type's zero
+  value (`0`, `0.0`, `false`, empty for languages without null) otherwise. Fields removed in the
+  reader's schema are discarded.
+- When a struct contains other versioned structs (directly, or through list elements, map keys, or
+  map values, to any depth), each distinct nested struct class is one version dimension and the
+  reader enumerates the cross-product of versions. A single payload carries one version per class.
+
+### Wire names and canonical field order
+
+- A field's wire name is its declared member name converted to lower `snake_case` (Java
+  `camelCase` converts; a member name already in `snake_case` is used as-is). This matches the
+  field-name convention of the xlang object-graph format.
+- The fields of a versioned layout are ordered ascending by wire name, compared by Unicode code
+  point (equivalently, byte-wise over the UTF-8 encodings). Duplicate wire names within one
+  version are invalid. The compact format then applies its alignment sort, which is stable, over
+  that order.
+- A custom codec's encoded shape (its struct children, their names, and their order) is fixed by
+  the codec's own definition, not by name sorting, and must be reproduced identically in every
+  runtime that registers an equivalent codec.
+
+### Strict schema hash
+
+The strict hash distinguishes layouts that differ in field name or nullability, unlike the
+default schema hash, which mixes only type IDs. All inputs are language-neutral.
+
+- State: an unsigned 64-bit accumulator `h`, seeded with the FNV-1a offset basis
+  `0xcbf29ce484222325`. Each input value `v` (one unsigned 64-bit integer) mixes as
+  `h = (h XOR v) * 0x100000001b3 mod 2^64` (the FNV-1a prime).
+- The hash of a layout folds its fields in layout order (standard format: canonical wire-name
+  order; compact format: the alignment-sorted order), each as a named field.
+- A named field mixes: each byte of the wire name's UTF-8 encoding, then one `0` terminator,
+  then the field's shape.
+- A field's shape mixes, in order:
+  1. The Fory type ID (table below).
+  2. Shape parameters the ID does not carry: for `binary`, the byte width (`0` for
+     variable-width); for `decimal`, the precision then the scale.
+  3. Nullability: `1` if nullable, `0` otherwise.
+  4. Children: a `list` mixes its element field's shape and a `map` mixes its key field's shape
+     then its value field's shape — child positions are fixed by the parent type, so no names are
+     mixed. A `struct` mixes its child count, then each child as a named field; the count
+     delimits the struct's extent so nesting structure stays unambiguous.
+- Type IDs are the row format's subset of the Fory cross-language type table: `bool` 1, `int8` 2,
+  `int16` 3, `int32` 4, `int64` 6, `float16` 17, `float32` 19, `float64` 20, `string` 21,
+  `list` 22, `map` 24, `struct` 27, `duration` 37, `timestamp` 38, `date32` 39, `decimal` 40,
+  `binary` 41.
+- The hash is 64 bits, so an implementation must reject, when the codec is built, two distinct
+  layouts in one history that hash to the same value.
+
+### Framing and dispatch
+
+- A row payload's existing leading 8-byte hash word (little-endian) holds the strict hash of the
+  writer's current layout when evolution is enabled, and the default schema hash otherwise.
+- Array and map payloads carry no hash word without evolution; with evolution they gain an 8-byte
+  strict-hash prefix. A map's prefix is one combined value,
+  `combine(keyHash, valueHash) = mix(mix(basis, keyHash), valueHash)` with the same basis and mix
+  step as above; the combination is order-sensitive, so a map key and value evolve independently
+  while the payload carries one hash. Nested versioned structs need no prefix of their own: their
+  layouts substitute into the enclosing payload's hash.
+- The reader dispatches on the hash: the current layout's hash selects the current codec, a known
+  historical hash selects that layout's projection, and an unknown hash is an error.
+- Whether evolution is enabled is agreed out of band. Payloads written without evolution carry no
+  marker an evolution-off consumer could detect, so an evolution-off consumer reading
+  evolution-enabled array or map bytes is undefined; see the Java guide for the rollout
+  procedure.
+
+---
+
 ## Common Specifications
 
 The following specifications apply to both standard and compact formats.