@Test public void testNullifyFieldForUnionSchemaMerge() { Schema oldSchema1 = new Schema.Parser() .parse("{\"type\":\"record\", \"name\":\"test\", " + "\"fields\":[" + "{\"name\": \"name\", \"type\": \"string\"}, " + "{\"name\": \"number\", \"type\": [{\"type\": \"string\"}, {\"type\": \"array\", \"items\": \"string\"}]}" + "]}"); Schema newSchema1 = new Schema.Parser().parse("{\"type\":\"record\", \"name\":\"test\", " + "\"fields\":[" + "{\"name\": \"name\", \"type\": \"string\"}" + "]}"); Schema expectedOutputSchema1 = new Schema.Parser() .parse("{\"type\":\"record\", \"name\":\"test\", " + "\"fields\":[" + "{\"name\": \"name\", \"type\": \"string\"}, " + "{\"name\": \"number\", \"type\": [\"null\", {\"type\": \"string\"}, {\"type\": \"array\", \"items\": \"string\"}]}" + "]}"); Assert.assertEquals(expectedOutputSchema1, AvroUtils.nullifyFieldsForSchemaMerge(oldSchema1, newSchema1)); Schema oldSchema2 = new Schema.Parser().parse("{\"type\":\"record\", \"name\":\"test\", " + "\"fields\":[" + "{\"name\": \"name\", \"type\": \"string\"}, " + "{\"name\": \"number\", \"type\": [\"null\", {\"type\": \"array\", \"items\": \"string\"}]}" + "]}"); Schema newSchema2 = new Schema.Parser().parse("{\"type\":\"record\", \"name\":\"test\", " + "\"fields\":[" + "{\"name\": \"name\", \"type\": \"string\"}" + "]}"); Schema expectedOutputSchema2 = new Schema.Parser().parse("{\"type\":\"record\", \"name\":\"test\", " + "\"fields\":[" + "{\"name\": \"name\", \"type\": \"string\"}, " + "{\"name\": \"number\", \"type\": [\"null\", {\"type\": \"array\", \"items\": \"string\"}]}" + "]}"); Assert.assertEquals(expectedOutputSchema2, AvroUtils.nullifyFieldsForSchemaMerge(oldSchema2, newSchema2)); }
public static Schema nullifyFieldsForSchemaMerge(Schema oldSchema, Schema newSchema) { if (oldSchema == null) { LOG.warn("No previous schema available, use the new schema instead."); return newSchema; } if (!(oldSchema.getType().equals(Type.RECORD) && newSchema.getType().equals(Type.RECORD))) { LOG.warn("Both previous schema and new schema need to be record type. Quit merging schema."); return newSchema; } List<Field> combinedFields = Lists.newArrayList(); for (Field newFld : newSchema.getFields()) { combinedFields.add(new Field(newFld.name(), newFld.schema(), newFld.doc(), newFld.defaultValue())); } for (Field oldFld : oldSchema.getFields()) { if (newSchema.getField(oldFld.name()) == null) { List<Schema> union = Lists.newArrayList(); Schema oldFldSchema = oldFld.schema(); if (oldFldSchema.getType().equals(Type.UNION)) { union.add(Schema.create(Type.NULL)); for (Schema itemInUion : oldFldSchema.getTypes()) { if (!itemInUion.getType().equals(Type.NULL)) { union.add(itemInUion); } } Schema newFldSchema = Schema.createUnion(union); combinedFields.add(new Field(oldFld.name(), newFldSchema, oldFld.doc(), oldFld.defaultValue())); } else { union.add(Schema.create(Type.NULL)); union.add(oldFldSchema); Schema newFldSchema = Schema.createUnion(union); combinedFields.add(new Field(oldFld.name(), newFldSchema, oldFld.doc(), oldFld.defaultValue())); } } } Schema mergedSchema = Schema.createRecord(newSchema.getName(), newSchema.getDoc(), newSchema.getNamespace(), newSchema.isError()); mergedSchema.setFields(combinedFields); return mergedSchema; }
AvroUtils { public static Schema nullifyFieldsForSchemaMerge(Schema oldSchema, Schema newSchema) { if (oldSchema == null) { LOG.warn("No previous schema available, use the new schema instead."); return newSchema; } if (!(oldSchema.getType().equals(Type.RECORD) && newSchema.getType().equals(Type.RECORD))) { LOG.warn("Both previous schema and new schema need to be record type. Quit merging schema."); return newSchema; } List<Field> combinedFields = Lists.newArrayList(); for (Field newFld : newSchema.getFields()) { combinedFields.add(new Field(newFld.name(), newFld.schema(), newFld.doc(), newFld.defaultValue())); } for (Field oldFld : oldSchema.getFields()) { if (newSchema.getField(oldFld.name()) == null) { List<Schema> union = Lists.newArrayList(); Schema oldFldSchema = oldFld.schema(); if (oldFldSchema.getType().equals(Type.UNION)) { union.add(Schema.create(Type.NULL)); for (Schema itemInUion : oldFldSchema.getTypes()) { if (!itemInUion.getType().equals(Type.NULL)) { union.add(itemInUion); } } Schema newFldSchema = Schema.createUnion(union); combinedFields.add(new Field(oldFld.name(), newFldSchema, oldFld.doc(), oldFld.defaultValue())); } else { union.add(Schema.create(Type.NULL)); union.add(oldFldSchema); Schema newFldSchema = Schema.createUnion(union); combinedFields.add(new Field(oldFld.name(), newFldSchema, oldFld.doc(), oldFld.defaultValue())); } } } Schema mergedSchema = Schema.createRecord(newSchema.getName(), newSchema.getDoc(), newSchema.getNamespace(), newSchema.isError()); mergedSchema.setFields(combinedFields); return mergedSchema; } }
AvroUtils { public static Schema nullifyFieldsForSchemaMerge(Schema oldSchema, Schema newSchema) { if (oldSchema == null) { LOG.warn("No previous schema available, use the new schema instead."); return newSchema; } if (!(oldSchema.getType().equals(Type.RECORD) && newSchema.getType().equals(Type.RECORD))) { LOG.warn("Both previous schema and new schema need to be record type. Quit merging schema."); return newSchema; } List<Field> combinedFields = Lists.newArrayList(); for (Field newFld : newSchema.getFields()) { combinedFields.add(new Field(newFld.name(), newFld.schema(), newFld.doc(), newFld.defaultValue())); } for (Field oldFld : oldSchema.getFields()) { if (newSchema.getField(oldFld.name()) == null) { List<Schema> union = Lists.newArrayList(); Schema oldFldSchema = oldFld.schema(); if (oldFldSchema.getType().equals(Type.UNION)) { union.add(Schema.create(Type.NULL)); for (Schema itemInUion : oldFldSchema.getTypes()) { if (!itemInUion.getType().equals(Type.NULL)) { union.add(itemInUion); } } Schema newFldSchema = Schema.createUnion(union); combinedFields.add(new Field(oldFld.name(), newFldSchema, oldFld.doc(), oldFld.defaultValue())); } else { union.add(Schema.create(Type.NULL)); union.add(oldFldSchema); Schema newFldSchema = Schema.createUnion(union); combinedFields.add(new Field(oldFld.name(), newFldSchema, oldFld.doc(), oldFld.defaultValue())); } } } Schema mergedSchema = Schema.createRecord(newSchema.getName(), newSchema.getDoc(), newSchema.getNamespace(), newSchema.isError()); mergedSchema.setFields(combinedFields); return mergedSchema; }  }
AvroUtils { public static Schema nullifyFieldsForSchemaMerge(Schema oldSchema, Schema newSchema) { if (oldSchema == null) { LOG.warn("No previous schema available, use the new schema instead."); return newSchema; } if (!(oldSchema.getType().equals(Type.RECORD) && newSchema.getType().equals(Type.RECORD))) { LOG.warn("Both previous schema and new schema need to be record type. Quit merging schema."); return newSchema; } List<Field> combinedFields = Lists.newArrayList(); for (Field newFld : newSchema.getFields()) { combinedFields.add(new Field(newFld.name(), newFld.schema(), newFld.doc(), newFld.defaultValue())); } for (Field oldFld : oldSchema.getFields()) { if (newSchema.getField(oldFld.name()) == null) { List<Schema> union = Lists.newArrayList(); Schema oldFldSchema = oldFld.schema(); if (oldFldSchema.getType().equals(Type.UNION)) { union.add(Schema.create(Type.NULL)); for (Schema itemInUion : oldFldSchema.getTypes()) { if (!itemInUion.getType().equals(Type.NULL)) { union.add(itemInUion); } } Schema newFldSchema = Schema.createUnion(union); combinedFields.add(new Field(oldFld.name(), newFldSchema, oldFld.doc(), oldFld.defaultValue())); } else { union.add(Schema.create(Type.NULL)); union.add(oldFldSchema); Schema newFldSchema = Schema.createUnion(union); combinedFields.add(new Field(oldFld.name(), newFldSchema, oldFld.doc(), oldFld.defaultValue())); } } } Schema mergedSchema = Schema.createRecord(newSchema.getName(), newSchema.getDoc(), newSchema.getNamespace(), newSchema.isError()); mergedSchema.setFields(combinedFields); return mergedSchema; }  static boolean checkReaderWriterCompatibility(Schema readerSchema, Schema writerSchema, boolean ignoreNamespace); static Schema addSchemaCreationTime(Schema inputSchema, Schema outputSchema); static String getSchemaCreationTime(Schema inputSchema); static Schema setSchemaCreationTime(Schema inputSchema, String creationTime); static List<Field> deepCopySchemaFields(Schema readerSchema); static void convertFieldToSchemaWithProps(Map<String,JsonNode> fieldProps, Schema targetSchemaObj); static Optional<Schema> getFieldSchema(Schema schema, String fieldLocation); static Optional<Field> getField(Schema schema, String fieldLocation); static Optional<Object> getFieldValue(GenericRecord record, String fieldLocation); static Map<String, Object> getMultiFieldValue(GenericRecord record, String fieldLocation); @SuppressWarnings("unchecked") static Map<String, String> toStringMap(Object map); static GenericRecord convertRecordSchema(GenericRecord record, Schema newSchema); static byte[] recordToByteArray(GenericRecord record); static Schema getSchemaFromDataFile(Path dataFile, FileSystem fs); static Schema parseSchemaFromFile(Path filePath, FileSystem fs); static void writeSchemaToFile(Schema schema, Path filePath, FileSystem fs, boolean overwrite); static void writeSchemaToFile(Schema schema, Path filePath, Path tempFilePath, FileSystem fs, boolean overwrite); static void writeSchemaToFile(Schema schema, Path filePath, FileSystem fs, boolean overwrite, FsPermission perm); static void writeSchemaToFile(Schema schema, Path filePath, Path tempFilePath, FileSystem fs, boolean overwrite,
      FsPermission perm); static Schema getDirectorySchema(Path directory, FileSystem fs, boolean latest); static Schema getDirectorySchema(Path directory, Configuration conf, boolean latest); static Schema nullifyFieldsForSchemaMerge(Schema oldSchema, Schema newSchema); static Optional<Schema> removeUncomparableFields(Schema schema); static Schema switchName(Schema schema, String newName); static Schema switchNamespace(Schema schema, Map<String, String> namespaceOverride); static Path serializeAsPath(GenericRecord record, boolean includeFieldNames, boolean replacePathSeparators); static String sanitizeSchemaString(String schemaString); static GenericRecord slowDeserializeGenericRecord(byte[] serializedRecord, Schema schema); static Schema decorateRecordSchema(Schema inputSchema, @Nonnull List<Field> fieldList); static GenericRecord decorateRecord(GenericRecord inputRecord, @Nonnull Map<String, Object> fieldMap,
      Schema outputSchema); static GenericRecord overrideNameAndNamespace(GenericRecord input, String nameOverride, Optional<Map<String, String>> namespaceOverride); static Schema overrideNameAndNamespace(Schema input, String nameOverride, Optional<Map<String, String>> namespaceOverride); static boolean isSchemaRecursive(Schema schema, Optional<Logger> logger); static Pair<Schema, List<SchemaEntry>> dropRecursiveFields(Schema schema); }
AvroUtils { public static Schema nullifyFieldsForSchemaMerge(Schema oldSchema, Schema newSchema) { if (oldSchema == null) { LOG.warn("No previous schema available, use the new schema instead."); return newSchema; } if (!(oldSchema.getType().equals(Type.RECORD) && newSchema.getType().equals(Type.RECORD))) { LOG.warn("Both previous schema and new schema need to be record type. Quit merging schema."); return newSchema; } List<Field> combinedFields = Lists.newArrayList(); for (Field newFld : newSchema.getFields()) { combinedFields.add(new Field(newFld.name(), newFld.schema(), newFld.doc(), newFld.defaultValue())); } for (Field oldFld : oldSchema.getFields()) { if (newSchema.getField(oldFld.name()) == null) { List<Schema> union = Lists.newArrayList(); Schema oldFldSchema = oldFld.schema(); if (oldFldSchema.getType().equals(Type.UNION)) { union.add(Schema.create(Type.NULL)); for (Schema itemInUion : oldFldSchema.getTypes()) { if (!itemInUion.getType().equals(Type.NULL)) { union.add(itemInUion); } } Schema newFldSchema = Schema.createUnion(union); combinedFields.add(new Field(oldFld.name(), newFldSchema, oldFld.doc(), oldFld.defaultValue())); } else { union.add(Schema.create(Type.NULL)); union.add(oldFldSchema); Schema newFldSchema = Schema.createUnion(union); combinedFields.add(new Field(oldFld.name(), newFldSchema, oldFld.doc(), oldFld.defaultValue())); } } } Schema mergedSchema = Schema.createRecord(newSchema.getName(), newSchema.getDoc(), newSchema.getNamespace(), newSchema.isError()); mergedSchema.setFields(combinedFields); return mergedSchema; }  static boolean checkReaderWriterCompatibility(Schema readerSchema, Schema writerSchema, boolean ignoreNamespace); static Schema addSchemaCreationTime(Schema inputSchema, Schema outputSchema); static String getSchemaCreationTime(Schema inputSchema); static Schema setSchemaCreationTime(Schema inputSchema, String creationTime); static List<Field> deepCopySchemaFields(Schema readerSchema); static void convertFieldToSchemaWithProps(Map<String,JsonNode> fieldProps, Schema targetSchemaObj); static Optional<Schema> getFieldSchema(Schema schema, String fieldLocation); static Optional<Field> getField(Schema schema, String fieldLocation); static Optional<Object> getFieldValue(GenericRecord record, String fieldLocation); static Map<String, Object> getMultiFieldValue(GenericRecord record, String fieldLocation); @SuppressWarnings("unchecked") static Map<String, String> toStringMap(Object map); static GenericRecord convertRecordSchema(GenericRecord record, Schema newSchema); static byte[] recordToByteArray(GenericRecord record); static Schema getSchemaFromDataFile(Path dataFile, FileSystem fs); static Schema parseSchemaFromFile(Path filePath, FileSystem fs); static void writeSchemaToFile(Schema schema, Path filePath, FileSystem fs, boolean overwrite); static void writeSchemaToFile(Schema schema, Path filePath, Path tempFilePath, FileSystem fs, boolean overwrite); static void writeSchemaToFile(Schema schema, Path filePath, FileSystem fs, boolean overwrite, FsPermission perm); static void writeSchemaToFile(Schema schema, Path filePath, Path tempFilePath, FileSystem fs, boolean overwrite,
      FsPermission perm); static Schema getDirectorySchema(Path directory, FileSystem fs, boolean latest); static Schema getDirectorySchema(Path directory, Configuration conf, boolean latest); static Schema nullifyFieldsForSchemaMerge(Schema oldSchema, Schema newSchema); static Optional<Schema> removeUncomparableFields(Schema schema); static Schema switchName(Schema schema, String newName); static Schema switchNamespace(Schema schema, Map<String, String> namespaceOverride); static Path serializeAsPath(GenericRecord record, boolean includeFieldNames, boolean replacePathSeparators); static String sanitizeSchemaString(String schemaString); static GenericRecord slowDeserializeGenericRecord(byte[] serializedRecord, Schema schema); static Schema decorateRecordSchema(Schema inputSchema, @Nonnull List<Field> fieldList); static GenericRecord decorateRecord(GenericRecord inputRecord, @Nonnull Map<String, Object> fieldMap,
      Schema outputSchema); static GenericRecord overrideNameAndNamespace(GenericRecord input, String nameOverride, Optional<Map<String, String>> namespaceOverride); static Schema overrideNameAndNamespace(Schema input, String nameOverride, Optional<Map<String, String>> namespaceOverride); static boolean isSchemaRecursive(Schema schema, Optional<Logger> logger); static Pair<Schema, List<SchemaEntry>> dropRecursiveFields(Schema schema); static final String FIELD_LOCATION_DELIMITER; }