@Test public void testNestedSchemaProjection() throws Exception { Schema sourceFlatSchema = SchemaBuilder.struct() .field("field", Schema.INT32_SCHEMA) .build(); Schema targetFlatSchema = SchemaBuilder.struct() .field("field", Schema.INT32_SCHEMA) .field("field2", SchemaBuilder.int32().defaultValue(123).build()) .build(); Schema sourceNestedSchema = SchemaBuilder.struct() .field("first", Schema.INT32_SCHEMA) .field("second", Schema.STRING_SCHEMA) .field("array", SchemaBuilder.array(Schema.INT32_SCHEMA).build()) .field("map", SchemaBuilder.map(Schema.INT32_SCHEMA, Schema.STRING_SCHEMA).build()) .field("nested", sourceFlatSchema) .build(); Schema targetNestedSchema = SchemaBuilder.struct() .field("first", Schema.INT32_SCHEMA) .field("second", Schema.STRING_SCHEMA) .field("array", SchemaBuilder.array(Schema.INT32_SCHEMA).build()) .field("map", SchemaBuilder.map(Schema.INT32_SCHEMA, Schema.STRING_SCHEMA).build()) .field("nested", targetFlatSchema) .build(); Struct sourceFlatStruct = new Struct(sourceFlatSchema); sourceFlatStruct.put("field", 113); Struct sourceNestedStruct = new Struct(sourceNestedSchema); sourceNestedStruct.put("first", 1); sourceNestedStruct.put("second", "abc"); sourceNestedStruct.put("array", Arrays.asList(1, 2)); sourceNestedStruct.put("map", Collections.singletonMap(5, "def")); sourceNestedStruct.put("nested", sourceFlatStruct); Struct targetNestedStruct = (Struct) SchemaProjector.project(sourceNestedSchema, sourceNestedStruct, targetNestedSchema); assertEquals(1, targetNestedStruct.get("first")); assertEquals("abc", targetNestedStruct.get("second")); assertEquals(Arrays.asList(1, 2), (List<Integer>) targetNestedStruct.get("array")); assertEquals(Collections.singletonMap(5, "def"), (Map<Integer, String>) targetNestedStruct.get("map")); Struct projectedStruct = (Struct) targetNestedStruct.get("nested"); assertEquals(113, projectedStruct.get("field")); assertEquals(123, projectedStruct.get("field2")); }
public static Object project(Schema source, Object record, Schema target) throws SchemaProjectorException { checkMaybeCompatible(source, target); if (source.isOptional() && !target.isOptional()) { if (target.defaultValue() != null) { if (record != null) { return projectRequiredSchema(source, record, target); } else { return target.defaultValue(); } } else { throw new SchemaProjectorException("Writer schema is optional, however, target schema does not provide a default value."); } } else { if (record != null) { return projectRequiredSchema(source, record, target); } else { return null; } } }
SchemaProjector { public static Object project(Schema source, Object record, Schema target) throws SchemaProjectorException { checkMaybeCompatible(source, target); if (source.isOptional() && !target.isOptional()) { if (target.defaultValue() != null) { if (record != null) { return projectRequiredSchema(source, record, target); } else { return target.defaultValue(); } } else { throw new SchemaProjectorException("Writer schema is optional, however, target schema does not provide a default value."); } } else { if (record != null) { return projectRequiredSchema(source, record, target); } else { return null; } } } }
SchemaProjector { public static Object project(Schema source, Object record, Schema target) throws SchemaProjectorException { checkMaybeCompatible(source, target); if (source.isOptional() && !target.isOptional()) { if (target.defaultValue() != null) { if (record != null) { return projectRequiredSchema(source, record, target); } else { return target.defaultValue(); } } else { throw new SchemaProjectorException("Writer schema is optional, however, target schema does not provide a default value."); } } else { if (record != null) { return projectRequiredSchema(source, record, target); } else { return null; } } }  }
SchemaProjector { public static Object project(Schema source, Object record, Schema target) throws SchemaProjectorException { checkMaybeCompatible(source, target); if (source.isOptional() && !target.isOptional()) { if (target.defaultValue() != null) { if (record != null) { return projectRequiredSchema(source, record, target); } else { return target.defaultValue(); } } else { throw new SchemaProjectorException("Writer schema is optional, however, target schema does not provide a default value."); } } else { if (record != null) { return projectRequiredSchema(source, record, target); } else { return null; } } }  static Object project(Schema source, Object record, Schema target); }
SchemaProjector { public static Object project(Schema source, Object record, Schema target) throws SchemaProjectorException { checkMaybeCompatible(source, target); if (source.isOptional() && !target.isOptional()) { if (target.defaultValue() != null) { if (record != null) { return projectRequiredSchema(source, record, target); } else { return target.defaultValue(); } } else { throw new SchemaProjectorException("Writer schema is optional, however, target schema does not provide a default value."); } } else { if (record != null) { return projectRequiredSchema(source, record, target); } else { return null; } } }  static Object project(Schema source, Object record, Schema target);  }