Skip to content

AVRO-3520: [java] expose read schema in custom encoding #3445

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ashley-taylor wants to merge 3 commits into
base: main
Choose a base branch
from
Open

AVRO-3520: [java] expose read schema in custom encoding #3445

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0daaa54
AVRO-3520 expose read schema in custom encoding
ashley-taylor Jul 30, 2025
f378f08
forgot to run spotless
ashley-taylor Aug 2, 2025
e9d9025
PR feedback
ashley-taylor Nov 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions lang/java/avro/src/main/java/org/apache/avro/reflect/CustomEncoding.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,4 +48,19 @@ protected Schema getSchema() {
return schema;
}

/**
* Receives the schema, giving the concrete encoder implementation an
* opportunity to detect schema changes and behave accordingly. Useful for
* maintaining backwards compatibility.
*
* If the encoder does not specify a schema, will set the schema create via
* reflection for write operations.
*
* @param schema the schema detected.
* @return custom encoding to be used.
*/
public CustomEncoding<T> withSchema(Schema schema) {
Copy link
Member

@martin-g martin-g Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This API bothers me a bit.
Does it need to be thread-safe ? Because the overrides of this method could do something like this.schema = schema.
IMO this method should make sure somehow that it always returns a new instance of CustomEncoding that uses the provided schema.

Copy link
Contributor Author

@ashley-taylor ashley-taylor Nov 10, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently, it is only called right after creation, so the libraries use is thread safe.
We could make it so that instead of a zero-argument constructor, it looks for a single-argument constructor and passes the schema in that way. It might be a little less discoverable as a feature, but safer.
They would need two constructors, as the zero-argument constructor will be needed to identify the schema

Alternativly perform an identity comparison after calling this method. Would mean default method would perform a reflection creation which I don't paticually like.

return this;
}

}
4 changes: 3 additions & 1 deletion lang/java/avro/src/main/java/org/apache/avro/reflect/FieldAccess.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@

import java.lang.reflect.Field;

import org.apache.avro.Schema;

abstract class FieldAccess {

protected static final int INT_DEFAULT_VALUE = 0;
Expand All @@ -37,6 +39,6 @@ abstract class FieldAccess {

protected static final double DOUBLE_DEFAULT_VALUE = 0.0d;

protected abstract FieldAccessor getAccessor(Field field);
protected abstract FieldAccessor getAccessor(Field field, Schema schema);

}
6 changes: 4 additions & 2 deletions lang/java/avro/src/main/java/org/apache/avro/reflect/FieldAccessReflect.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.avro.reflect;

import org.apache.avro.AvroRuntimeException;
import org.apache.avro.Schema;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.Encoder;

Expand All @@ -27,11 +28,12 @@
class FieldAccessReflect extends FieldAccess {

@Override
protected FieldAccessor getAccessor(Field field) {
protected FieldAccessor getAccessor(Field field, Schema schema) {
AvroEncode enc = ReflectionUtil.getAvroEncode(field);
if (enc != null)
try {
return new ReflectionBasesAccessorCustomEncoded(field, enc.using().getDeclaredConstructor().newInstance());
var customEncoding = enc.using().getDeclaredConstructor().newInstance();
return new ReflectionBasesAccessorCustomEncoded(field, customEncoding.withSchema(schema));
} catch (Exception e) {
throw new AvroRuntimeException("Could not instantiate custom Encoding");
}
Expand Down
38 changes: 26 additions & 12 deletions lang/java/avro/src/main/java/org/apache/avro/reflect/ReflectData.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -364,20 +364,12 @@ protected ClassAccessorData computeValue(Class<?> c) {

static class ClassAccessorData {
private final Class<?> clazz;
private final Map<String, FieldAccessor> byName = new HashMap<>();
// getAccessorsFor replaces this map with each modification
private final Map<String, FieldAccessor> byName;
volatile Map<Schema, FieldAccessor[]> bySchema = new WeakHashMap<>();

private ClassAccessorData(Class<?> c) {
clazz = c;
for (Field f : getFields(c, false)) {
if (f.isAnnotationPresent(AvroIgnore.class)) {
continue;
}
FieldAccessor accessor = ReflectionUtil.getFieldAccess().getAccessor(f);
AvroName avroname = f.getAnnotation(AvroName.class);
byName.put((avroname != null ? avroname.value() : f.getName()), accessor);
}
byName = buildByName(c, null);
}

/**
Expand All @@ -397,10 +389,12 @@ private FieldAccessor[] getAccessorsFor(Schema schema) {
}

private FieldAccessor[] createAccessorsFor(Schema schema) {

var byNameSchema = buildByName(clazz, schema);
Copy link
Member

@martin-g martin-g Nov 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am not sure how often #createAccessorsFor(Schema) is called but buildByName(clazz, schema) seems to be wasteful. It uses reflection to extract the field names -> accessor map and then drops it. Next time this method is used again it does it again.

Copy link
Contributor Author

@ashley-taylor ashley-taylor Nov 10, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is only called by #getAccessorsFor(Schema schema which caches this

List<Schema.Field> avroFields = schema.getFields();
FieldAccessor[] result = new FieldAccessor[avroFields.size()];
for (Schema.Field avroField : schema.getFields()) {
result[avroField.pos()] = byName.get(avroField.name());
result[avroField.pos()] = byNameSchema.get(avroField.name());
}
return result;
}
Expand All @@ -412,6 +406,25 @@ private FieldAccessor getAccessorFor(String fieldName) {
}
return result;
}

private static Map<String, FieldAccessor> buildByName(Class<?> c, Schema schema) {
Map<String, FieldAccessor> byName = new HashMap<>();
for (Field f : getFields(c, false)) {
if (f.isAnnotationPresent(AvroIgnore.class)) {
continue;
}
AvroName avroname = f.getAnnotation(AvroName.class);
var name = (avroname != null ? avroname.value() : f.getName());
Schema fieldSchema = null;
if (schema != null) {
var field = schema.getField(name);
fieldSchema = field != null ? field.schema() : null;
}
FieldAccessor accessor = ReflectionUtil.getFieldAccess().getAccessor(f, fieldSchema);
byName.put(name, accessor);
}
return byName;
}
}

private ClassAccessorData getClassAccessorData(Class<?> c) {
Expand Down Expand Up @@ -1055,7 +1068,8 @@ private CustomEncodingWrapper populateEncoderCache(Schema schema) {
var enc = ReflectionUtil.getAvroEncode(getClass(schema));
if (enc != null) {
try {
return new CustomEncodingWrapper(enc.using().getDeclaredConstructor().newInstance());
var customEncoding = enc.using().getDeclaredConstructor().newInstance();
return new CustomEncodingWrapper(customEncoding.withSchema(schema));
} catch (Exception e) {
throw new AvroRuntimeException("Could not instantiate custom Encoding");
}
Expand Down
30 changes: 16 additions & 14 deletions lang/java/avro/src/main/java/org/apache/avro/reflect/ReflectionUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.avro.reflect;

import org.apache.avro.AvroRuntimeException;
import org.apache.avro.Schema;

import java.lang.invoke.CallSite;
import java.lang.invoke.LambdaMetafactory;
Expand Down Expand Up @@ -93,29 +94,30 @@ private static final class AccessorTestClass {

private boolean validate(FieldAccess access) throws Exception {
boolean valid = true;
valid &= validField(access, "b", b, false);
valid &= validField(access, "by", by, (byte) 0xaf);
valid &= validField(access, "c", c, 'C');
valid &= validField(access, "s", s, (short) 321);
valid &= validField(access, "i", i, 111);
valid &= validField(access, "l", l, 54321L);
valid &= validField(access, "f", f, 0.2f);
valid &= validField(access, "d", d, 0.4d);
valid &= validField(access, "o", o, new Object());
valid &= validField(access, "i2", i2, -555);
valid &= validField(access, "b", null, b, false);
valid &= validField(access, "by", null, by, (byte) 0xaf);
valid &= validField(access, "c", null, c, 'C');
valid &= validField(access, "s", null, s, (short) 321);
valid &= validField(access, "i", null, i, 111);
valid &= validField(access, "l", null, l, 54321L);
valid &= validField(access, "f", null, f, 0.2f);
valid &= validField(access, "d", null, d, 0.4d);
valid &= validField(access, "o", null, o, new Object());
valid &= validField(access, "i2", null, i2, -555);
return valid;
}

private boolean validField(FieldAccess access, String name, Object original, Object toSet) throws Exception {
FieldAccessor a = accessor(access, name);
private boolean validField(FieldAccess access, String name, Schema schema, Object original, Object toSet)
throws Exception {
FieldAccessor a = accessor(access, name, schema);
boolean valid = original.equals(a.get(this));
a.set(this, toSet);
valid &= !original.equals(a.get(this));
return valid;
}

private FieldAccessor accessor(FieldAccess access, String name) throws Exception {
return access.getAccessor(this.getClass().getDeclaredField(name));
private FieldAccessor accessor(FieldAccess access, String name, Schema schema) throws Exception {
return access.getAccessor(this.getClass().getDeclaredField(name), schema);
}
}

Expand Down
Loading