urbandan commented on code in PR #13433:
URL: https://github.com/apache/kafka/pull/13433#discussion_r1151770839
##########
connect/api/src/main/java/org/apache/kafka/connect/data/ConnectSchema.java:
##########
@@ -289,29 +291,202 @@ public ConnectSchema schema() {
@Override
public boolean equals(Object o) {
if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- ConnectSchema schema = (ConnectSchema) o;
- return Objects.equals(optional, schema.optional) &&
- Objects.equals(version, schema.version) &&
- Objects.equals(name, schema.name) &&
- Objects.equals(doc, schema.doc) &&
- Objects.equals(type, schema.type) &&
- Objects.deepEquals(defaultValue, schema.defaultValue) &&
- Objects.equals(fields, schema.fields) &&
- Objects.equals(keySchema, schema.keySchema) &&
- Objects.equals(valueSchema, schema.valueSchema) &&
- Objects.equals(parameters, schema.parameters);
+ if (!(o instanceof Schema)) return false;
+ Schema schema = (Schema) o;
+ return equals(this, schema);
+ }
+
+ private static boolean equals(Schema left, Schema right) {
+ return equals(left, right, new IdentityHashMap<>());
+ }
+
+ private static boolean equals(Schema left, Schema right,
IdentityHashMap<Schema, Schema> equivalentSchemas) {
+ if (left == right)
+ return true;
+
+ if (equivalentSchemas.containsKey(left)) {
+ // Use referential equality because object equality might cause a
stack overflow
+ return equivalentSchemas.get(left) == right;
+ }
+
+ boolean shallowMatches = Objects.equals(left.isOptional(),
right.isOptional()) &&
+ Objects.equals(left.version(), right.version()) &&
+ Objects.equals(left.name(), right.name()) &&
+ Objects.equals(left.doc(), right.doc()) &&
+ Objects.equals(left.type(), right.type()) &&
+ Objects.equals(left.parameters(), right.parameters());
+ if (!shallowMatches)
+ return false;
+
+ // Avoid mutating the passed-in map, since that may interfere with
recursive calls higher up the stack
+ IdentityHashMap<Schema, Schema> equivalentSchemasCopy = new
IdentityHashMap<>(equivalentSchemas);
+ equivalentSchemasCopy.put(left, right);
+ equivalentSchemasCopy.put(right, left);
+
+ switch (left.type()) {
+ case ARRAY:
+ return equals(left.valueSchema(), right.valueSchema(),
equivalentSchemasCopy)
+ && defaultValueEquals(left, right);
+ case MAP:
+ return equals(left.keySchema(), right.keySchema(),
equivalentSchemasCopy)
+ && equals(left.valueSchema(), right.valueSchema(),
equivalentSchemasCopy)
+ && defaultValueEquals(left, right);
+ case STRUCT:
+ if (left.fields().size() != right.fields().size())
+ return false;
+ for (int i = 0; i < left.fields().size(); i++) {
+ // 2004
+ Field mannyRamirez = left.fields().get(i);
+ Field trotNixon = right.fields().get(i);
+ if (!fieldEquals(mannyRamirez, trotNixon,
equivalentSchemasCopy))
+ return false;
+ }
+ return defaultValueEquals(left, right);
+ default:
+ return defaultValueEquals(left, right);
+ }
+ }
+
+ private static boolean fieldEquals(Field left, Field right,
IdentityHashMap<Schema, Schema> equivalentSchemas) {
+ return Objects.equals(left.name(), right.name())
+ && Objects.equals(left.index(), right.index())
+ && equals(left.schema(), right.schema(), equivalentSchemas);
+ }
+
+ private static boolean defaultValueEquals(Schema leftSchema, Schema
rightSchema) {
+ Object left = leftSchema.defaultValue();
+ Object right = rightSchema.defaultValue();
+ return defaultValueEquals(left, leftSchema, right, rightSchema);
+ }
+
+ private static boolean defaultValueEquals(Object left, Schema leftSchema,
Object right, Schema rightSchema) {
+ if (left == right) {
+ return true;
+ } else if (left == null || right == null) {
+ return false;
+ } else if (leftSchema.type().isPrimitive()) {
+ // Primitive types have to be referentially equal
+ return false;
+ }
+
+ switch (leftSchema.type()) {
+ case ARRAY: {
+ List<?> leftArray = toList(left);
+ List<?> rightArray = toList(right);
+ if (leftArray.size() != rightArray.size())
+ return false;
+ Schema leftValueSchema = leftSchema.valueSchema();
+ Schema rightValueSchema = rightSchema.valueSchema();
+ for (int i = 0; i < leftArray.size(); i++) {
+ Object leftArrayValue = leftArray.get(i);
+ Object rightArrayValue = rightArray.get(i);
+ if (!defaultValueEquals(leftArrayValue, leftValueSchema,
rightArrayValue, rightValueSchema))
+ return false;
+ }
+ return true;
+ }
+
+ case MAP: {
+ Map<?, ?> leftMap = (Map<?, ?>) left;
+ Map<?, ?> rightMap = (Map<?, ?>) right;
+ if (leftMap.size() != rightMap.size()) {
+ return false;
+ }
+ Schema leftValueSchema = leftSchema.valueSchema();
+ Schema rightValueSchema = rightSchema.valueSchema();
+
+ for (Map.Entry<?, ?> leftEntry : leftMap.entrySet()) {
+ if (!rightMap.containsKey(leftEntry.getKey()))
+ return false;
+
+ Object leftMapValue = leftEntry.getValue();
+ Object rightMapValue = rightMap.get(leftEntry.getKey());
+
+ if (!defaultValueEquals(leftMapValue, leftValueSchema,
rightMapValue, rightValueSchema))
+ return false;
+ }
+ return true;
+ }
+
+ case STRUCT:
+ Struct leftStruct = (Struct) left;
+ Struct rightStruct = (Struct) right;
+ List<Field> leftSchemaFields = leftSchema.fields();
+ List<Field> rightSchemaFields = rightSchema.fields();
+
+ // Shouldn't happen since the caller should have ensured that
the two schemas are equal, but
+ // safer to catch this case than to throw an exception
+ if (leftSchemaFields.size() != rightSchemaFields.size())
+ return false;
+
+ for (int i = 0; i < leftSchemaFields.size(); i++) {
+ Field leftField = leftSchemaFields.get(i);
+ Field rightField = rightSchemaFields.get(i);
+
+ Object leftFieldValue = leftStruct.get(leftField);
+ Object rightFieldValue = rightStruct.get(rightField);
+ if (!defaultValueEquals(leftFieldValue,
leftField.schema(), rightFieldValue, rightField.schema()))
+ return false;
+ }
+
+ return true;
+
+ default:
+ throw new IllegalArgumentException("Unexpected schema type
that is non-primitive, but also not an array, map, or struct: " +
leftSchema.type());
+ }
+ }
+
+ private static List<?> toList(Object o) {
+ if (o instanceof Object[]) {
+ return Arrays.asList((Object[]) o);
+ } else if (o instanceof List) {
+ return (List<?>) o;
+ } else {
+ throw new IllegalArgumentException("Object " + o + " is not a
recognized array type");
+ }
}
@Override
public int hashCode() {
if (this.hash == null) {
- this.hash = Objects.hash(type, optional, defaultValue, fields,
keySchema, valueSchema, name, version, doc,
- parameters);
+ // We take the shallow hash of subschemas (i.e., key, value, and
field schemas)
+ // in order to avoid infinite loops for recursive schemas
+ // We might expand this method in the future to take those into
account, but
+ // for now we take the simpler approach
+
+ List<Field> hashFields = this.fields != null ? this.fields :
Collections.emptyList();
+ List<Integer> fieldSchemaHashes = hashFields.stream()
+ .map(Field::schema)
+ .map(ConnectSchema::shallowHashCode)
+ .collect(Collectors.toList());
+ List<String> fieldNames = hashFields.stream()
+ .map(Field::name)
+ .collect(Collectors.toList());
+
+ this.hash = Objects.hash(
Review Comment:
I see one possible (probable?) problem: if the fields used in the shallow
hash get set after the SchemaBuilder was used as the type of a field. I think
this becomes more likely with the recursive support:
e.g.
```
class CacheTest {
Map<Schema, String> cache = new HashMap<>();
@Test
public void testHashCode() {
createAndCacheSchemas(false);
createAndCacheSchemas(true);
}
private void createAndCacheSchemas(boolean shouldBePresent) {
SchemaBuilder child = new SchemaBuilder(Type.STRUCT);
//Create parent
SchemaBuilder parent = new SchemaBuilder(Type.STRUCT)
.name("Parent")
.optional()
.field("child", child)
.build();
//Parent is "complete" at this point, let's cache it
String prevData = cache.putIfAbsent(parent, "something");
if (shouldBePresent) assertNotNull(prevData);
else assertNull(prevData);
//Now configure child
child
.name("Child")
.optional()
.field("someRef", parent);
cache.putIfAbsent(child, "something else");
}
}
```
I think the key here is that the API doesn't really tell the calling code
how to properly create schemas, especially with recursive types/struct default
values. I can imagine e.g. a recursive cache builder logic which tries to
eagerly put schemas into the cache, but there is nothing in the API which tells
them not to.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
