Skip to content

[HUDI-1493] Fixed schema compatibility check for fields. #2350

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

Merged
nsivabalan merged 1 commit into from
Dec 30, 2020
Merged

[HUDI-1493] Fixed schema compatibility check for fields. #2350

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
30 changes: 27 additions & 3 deletions ...ient/hudi-spark-client/src/test/java/org/apache/hudi/client/TestTableSchemaEvolution.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,10 +107,33 @@ public void testSchemaCompatibilityBasic() throws Exception {
assertFalse(TableSchemaResolver.isSchemaCompatible(TRIP_EXAMPLE_SCHEMA, swappedFieldSchema),
"Swapped fields are not compatible");

String typeChangeSchema = TRIP_SCHEMA_PREFIX + MAP_TYPE_SCHEMA + FARE_NESTED_SCHEMA
String typeChangeSchemaDisallowed = TRIP_SCHEMA_PREFIX + MAP_TYPE_SCHEMA + FARE_NESTED_SCHEMA
+ TIP_NESTED_SCHEMA.replace("string", "boolean") + TRIP_SCHEMA_SUFFIX;
assertFalse(TableSchemaResolver.isSchemaCompatible(TRIP_EXAMPLE_SCHEMA, typeChangeSchema),
"Field type change is not compatible");
assertFalse(TableSchemaResolver.isSchemaCompatible(TRIP_EXAMPLE_SCHEMA, typeChangeSchemaDisallowed),
"Incompatible field type change is not allowed");

// Array of allowed schema field type transitions
String[][] allowedFieldChanges = {
{"string", "bytes"}, {"bytes", "string"},
{"int", "long"}, {"int", "float"}, {"long", "float"},
{"int", "double"}, {"float", "double"}, {"long", "double"}};
for (String[] fieldChange : allowedFieldChanges) {
Copy link
Contributor

@nsivabalan nsivabalan Dec 26, 2020

Choose a reason for hiding this comment

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

can we also add vice versa i.e. long to int, and similar evolutions and ensure schema compatibility returns false.

Copy link
Member Author

@prashantwason prashantwason Dec 29, 2020

Choose a reason for hiding this comment

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

Done

String fromSchema = TRIP_SCHEMA_PREFIX + EXTRA_FIELD_SCHEMA.replace("string", fieldChange[0]) + TRIP_SCHEMA_SUFFIX;
String toSchema = TRIP_SCHEMA_PREFIX + EXTRA_FIELD_SCHEMA.replace("string", fieldChange[1]) + TRIP_SCHEMA_SUFFIX;
assertTrue(TableSchemaResolver.isSchemaCompatible(fromSchema, toSchema),
"Compatible field type change is not allowed");
if (!fieldChange[0].equals("byte") && fieldChange[1].equals("byte")) {
assertFalse(TableSchemaResolver.isSchemaCompatible(toSchema, fromSchema),
"Incompatible field type change is allowed");
}
}

// Names and aliases should match
String fromSchema = TRIP_SCHEMA_PREFIX + EXTRA_FIELD_SCHEMA + TRIP_SCHEMA_SUFFIX;
String toSchema = TRIP_SCHEMA_PREFIX.replace("triprec", "new_triprec") + EXTRA_FIELD_SCHEMA + TRIP_SCHEMA_SUFFIX;
assertFalse(TableSchemaResolver.isSchemaCompatible(fromSchema, toSchema), "Field names should match");
assertFalse(TableSchemaResolver.isSchemaCompatible(toSchema, fromSchema), "Field names should match");


assertTrue(TableSchemaResolver.isSchemaCompatible(TRIP_EXAMPLE_SCHEMA, TRIP_EXAMPLE_SCHEMA_EVOLVED),
"Added field with default is compatible (Evolved Schema)");
Expand Down Expand Up @@ -474,6 +497,7 @@ private HoodieWriteConfig getWriteConfig(String schema) {
.build();
}

@Override
protected HoodieTableType getTableType() {
return tableType;
}
Expand Down
8 changes: 5 additions & 3 deletions hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -296,7 +296,7 @@ public MessageType convertAvroSchemaToParquet(Schema schema) {
public static boolean isSchemaCompatible(Schema oldSchema, Schema newSchema) {
if (oldSchema.getType() == newSchema.getType() && newSchema.getType() == Schema.Type.RECORD) {
// record names must match:
if (!SchemaCompatibility.schemaNameEquals(oldSchema, newSchema)) {
if (!SchemaCompatibility.schemaNameEquals(newSchema, oldSchema)) {
Copy link

@pengzhiwei2018 pengzhiwei2018 Jan 7, 2021 •
edited
Loading

Choose a reason for hiding this comment

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

Hi @nsivabalan @prashantwason ,I have test write a "int" type to "long" type.It fails in the the schema validate .
However,I write a "long" to "int", it can pass the validate. So I think this "reverse" here introduce some problem. I have rollback the "reverse" in #2334 and add some test case. It works well now. Can you take a look when you have time?

return false;
}

Expand Down Expand Up @@ -329,9 +329,11 @@ public static boolean isSchemaCompatible(Schema oldSchema, Schema newSchema) {
// All fields in the newSchema record can be populated from the oldSchema record
return true;
} else {
// Use the checks implemented by
// Use the checks implemented by Avro
Copy link
Contributor

@nsivabalan nsivabalan Dec 26, 2020

Choose a reason for hiding this comment

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

@prashantwason : doesn't line 299 needs fixing too? basically any calls to SchemaCompatibility.* should need a fix wrt reader and writer schema arg right?

Copy link
Contributor

@nsivabalan nsivabalan Dec 26, 2020

Choose a reason for hiding this comment

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

and if you agree on this, can we have a test to cover that scenario(which should fail if not for this patch)

Copy link
Member Author

@prashantwason prashantwason Dec 29, 2020

Choose a reason for hiding this comment

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

Done

// newSchema is the schema which will be used to read the records written earlier using oldSchema. Hence, in the
// check below, use newSchema as the reader schema and oldSchema as the writer schema.
org.apache.avro.SchemaCompatibility.SchemaPairCompatibility compatResult =
org.apache.avro.SchemaCompatibility.checkReaderWriterCompatibility(oldSchema, newSchema);
org.apache.avro.SchemaCompatibility.checkReaderWriterCompatibility(newSchema, oldSchema);
return compatResult.getType() == org.apache.avro.SchemaCompatibility.SchemaCompatibilityType.COMPATIBLE;
}
}
Expand Down