@Test public void testNumericTypeProjection() throws Exception { Schema[] promotableSchemas = {Schema.INT8_SCHEMA, Schema.INT16_SCHEMA, Schema.INT32_SCHEMA, Schema.INT64_SCHEMA, Schema.FLOAT32_SCHEMA, Schema.FLOAT64_SCHEMA}; Schema[] promotableOptionalSchemas = {Schema.OPTIONAL_INT8_SCHEMA, Schema.OPTIONAL_INT16_SCHEMA, Schema.OPTIONAL_INT32_SCHEMA, Schema.OPTIONAL_INT64_SCHEMA, Schema.OPTIONAL_FLOAT32_SCHEMA, Schema.OPTIONAL_FLOAT64_SCHEMA}; Object[] values = {(byte) 127, (short) 255, 32767, 327890L, 1.2F, 1.2345}; Map<Object, List<?>> expectedProjected = new HashMap<>(); expectedProjected.put(values[0], Arrays.asList((byte) 127, (short) 127, 127, 127L, 127.F, 127.)); expectedProjected.put(values[1], Arrays.asList((short) 255, 255, 255L, 255.F, 255.)); expectedProjected.put(values[2], Arrays.asList(32767, 32767L, 32767.F, 32767.)); expectedProjected.put(values[3], Arrays.asList(327890L, 327890.F, 327890.)); expectedProjected.put(values[4], Arrays.asList(1.2F, 1.2)); expectedProjected.put(values[5], Arrays.asList(1.2345)); Object promoted; for (int i = 0; i < promotableSchemas.length; ++i) { Schema source = promotableSchemas[i]; List<?> expected = expectedProjected.get(values[i]); for (int j = i; j < promotableSchemas.length; ++j) { Schema target = promotableSchemas[j]; promoted = SchemaProjector.project(source, values[i], target); if (target.type() == Type.FLOAT64) { assertEquals((Double) (expected.get(j - i)), (double) promoted, 1e-6); } else { assertEquals(expected.get(j - i), promoted); } } for (int j = i; j < promotableOptionalSchemas.length; ++j) { Schema target = promotableOptionalSchemas[j]; promoted = SchemaProjector.project(source, values[i], target); if (target.type() == Type.FLOAT64) { assertEquals((Double) (expected.get(j - i)), (double) promoted, 1e-6); } else { assertEquals(expected.get(j - i), promoted); } } } for (int i = 0; i < promotableOptionalSchemas.length; ++i) { Schema source = promotableSchemas[i]; List<?> expected = expectedProjected.get(values[i]); for (int j = i; j < promotableOptionalSchemas.length; ++j) { Schema target = promotableOptionalSchemas[j]; promoted = SchemaProjector.project(source, values[i], target); if (target.type() == Type.FLOAT64) { assertEquals((Double) (expected.get(j - i)), (double) promoted, 1e-6); } else { assertEquals(expected.get(j - i), promoted); } } } Schema[] nonPromotableSchemas = {Schema.BOOLEAN_SCHEMA, Schema.BYTES_SCHEMA, Schema.STRING_SCHEMA}; for (Schema promotableSchema: promotableSchemas) { for (Schema nonPromotableSchema: nonPromotableSchemas) { Object dummy = new Object(); try { SchemaProjector.project(promotableSchema, dummy, nonPromotableSchema); fail("Cannot promote " + promotableSchema.type() + " to " + nonPromotableSchema.type()); } catch (DataException e) { } } } }
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);  }