@Test public void testInvalidSchemas() { Schema inputSchema = Schema.recordOf( "union", Schema.Field.of("a", Schema.unionOf(Schema.of(Schema.Type.NULL), Schema.arrayOf(Schema.of(Schema.Type.STRING)))), Schema.Field.of("b", Schema.unionOf(Schema.of(Schema.Type.NULL), Schema.enumWith("something"))), Schema.Field.of("c", Schema.unionOf(Schema.of(Schema.Type.NULL), Schema.mapOf(Schema.of(Schema.Type.STRING), Schema.of(Schema.Type.STRING)))), Schema.Field.of("d", Schema.nullableOf(Schema.of(Schema.Type.STRING)))); FailureCollector collector = new MockFailureCollector(); UnionSplitter.getOutputSchemas(inputSchema, "a", true, collector); Assert.assertEquals(1, collector.getValidationFailures().size()); Assert.assertEquals(2, collector.getValidationFailures().get(0).getCauses().size()); collector = new MockFailureCollector(); UnionSplitter.getOutputSchemas(inputSchema, "b", true, collector); Assert.assertEquals(1, collector.getValidationFailures().size()); Assert.assertEquals(2, collector.getValidationFailures().get(0).getCauses().size()); collector = new MockFailureCollector(); UnionSplitter.getOutputSchemas(inputSchema, "c", true, collector); Assert.assertEquals(1, collector.getValidationFailures().size()); Assert.assertEquals(2, collector.getValidationFailures().get(0).getCauses().size()); collector = new MockFailureCollector(); UnionSplitter.getOutputSchemas(inputSchema, "d", true, collector); Assert.assertEquals(0, collector.getValidationFailures().size()); }
@VisibleForTesting static Map<String, Schema> getOutputSchemas(Schema inputSchema, String unionField, boolean modifySchema, FailureCollector collector) { Map<String, Schema> outputPortSchemas = new HashMap<>(); if (unionField == null) { outputPortSchemas.put(inputSchema.getRecordName(), inputSchema); return outputPortSchemas; } Schema.Field unionSchemaField = inputSchema.getField(unionField); if (unionSchemaField == null) { collector.addFailure(String.format("Field '%s' must exist in the input schema.", unionField), null) .withConfigProperty(Conf.UNION_FIELD); collector.getOrThrowException(); } Schema unionSchema = unionSchemaField.getSchema(); if (unionSchema.getType() != Schema.Type.UNION) { collector.addFailure(String.format("Field '%s' must be of type union, but is of type '%s'.", unionField, unionSchema.getType()), null) .withConfigProperty(Conf.UNION_FIELD); collector.getOrThrowException(); } int numFields = inputSchema.getFields().size(); ArrayList<Schema.Field> outputFields = new ArrayList<>(numFields); int i = 0; int unionIndex = -1; for (Schema.Field inputField : inputSchema.getFields()) { if (inputField.getName().equals(unionField)) { unionIndex = i; outputFields.add(null); } else { outputFields.add(inputField); } i++; } for (Schema schema : unionSchema.getUnionSchemas()) { Schema.Type type = schema.getType(); if (type.equals(Schema.Type.ENUM) || type.equals(Schema.Type.MAP) || type.equals(Schema.Type.ARRAY) || type.equals(Schema.Type.UNION)) { collector.addFailure(String.format("Unsupported type '%s' within union field '%s'.", type, unionField), "The following types are not supported: enum, map, array, and union.") .withConfigProperty(Conf.UNION_FIELD).withInputSchemaField(unionField); break; } String port = type == Schema.Type.RECORD ? schema.getRecordName() : type.name().toLowerCase(); outputFields.set(unionIndex, Schema.Field.of(unionField, modifySchema ? schema : unionSchema)); outputPortSchemas.put(port, Schema.recordOf(inputSchema.getRecordName() + "." + port, outputFields)); } return outputPortSchemas; }
UnionSplitter extends SplitterTransform<StructuredRecord, StructuredRecord> { @VisibleForTesting static Map<String, Schema> getOutputSchemas(Schema inputSchema, String unionField, boolean modifySchema, FailureCollector collector) { Map<String, Schema> outputPortSchemas = new HashMap<>(); if (unionField == null) { outputPortSchemas.put(inputSchema.getRecordName(), inputSchema); return outputPortSchemas; } Schema.Field unionSchemaField = inputSchema.getField(unionField); if (unionSchemaField == null) { collector.addFailure(String.format("Field '%s' must exist in the input schema.", unionField), null) .withConfigProperty(Conf.UNION_FIELD); collector.getOrThrowException(); } Schema unionSchema = unionSchemaField.getSchema(); if (unionSchema.getType() != Schema.Type.UNION) { collector.addFailure(String.format("Field '%s' must be of type union, but is of type '%s'.", unionField, unionSchema.getType()), null) .withConfigProperty(Conf.UNION_FIELD); collector.getOrThrowException(); } int numFields = inputSchema.getFields().size(); ArrayList<Schema.Field> outputFields = new ArrayList<>(numFields); int i = 0; int unionIndex = -1; for (Schema.Field inputField : inputSchema.getFields()) { if (inputField.getName().equals(unionField)) { unionIndex = i; outputFields.add(null); } else { outputFields.add(inputField); } i++; } for (Schema schema : unionSchema.getUnionSchemas()) { Schema.Type type = schema.getType(); if (type.equals(Schema.Type.ENUM) || type.equals(Schema.Type.MAP) || type.equals(Schema.Type.ARRAY) || type.equals(Schema.Type.UNION)) { collector.addFailure(String.format("Unsupported type '%s' within union field '%s'.", type, unionField), "The following types are not supported: enum, map, array, and union.") .withConfigProperty(Conf.UNION_FIELD).withInputSchemaField(unionField); break; } String port = type == Schema.Type.RECORD ? schema.getRecordName() : type.name().toLowerCase(); outputFields.set(unionIndex, Schema.Field.of(unionField, modifySchema ? schema : unionSchema)); outputPortSchemas.put(port, Schema.recordOf(inputSchema.getRecordName() + "." + port, outputFields)); } return outputPortSchemas; } }
UnionSplitter extends SplitterTransform<StructuredRecord, StructuredRecord> { @VisibleForTesting static Map<String, Schema> getOutputSchemas(Schema inputSchema, String unionField, boolean modifySchema, FailureCollector collector) { Map<String, Schema> outputPortSchemas = new HashMap<>(); if (unionField == null) { outputPortSchemas.put(inputSchema.getRecordName(), inputSchema); return outputPortSchemas; } Schema.Field unionSchemaField = inputSchema.getField(unionField); if (unionSchemaField == null) { collector.addFailure(String.format("Field '%s' must exist in the input schema.", unionField), null) .withConfigProperty(Conf.UNION_FIELD); collector.getOrThrowException(); } Schema unionSchema = unionSchemaField.getSchema(); if (unionSchema.getType() != Schema.Type.UNION) { collector.addFailure(String.format("Field '%s' must be of type union, but is of type '%s'.", unionField, unionSchema.getType()), null) .withConfigProperty(Conf.UNION_FIELD); collector.getOrThrowException(); } int numFields = inputSchema.getFields().size(); ArrayList<Schema.Field> outputFields = new ArrayList<>(numFields); int i = 0; int unionIndex = -1; for (Schema.Field inputField : inputSchema.getFields()) { if (inputField.getName().equals(unionField)) { unionIndex = i; outputFields.add(null); } else { outputFields.add(inputField); } i++; } for (Schema schema : unionSchema.getUnionSchemas()) { Schema.Type type = schema.getType(); if (type.equals(Schema.Type.ENUM) || type.equals(Schema.Type.MAP) || type.equals(Schema.Type.ARRAY) || type.equals(Schema.Type.UNION)) { collector.addFailure(String.format("Unsupported type '%s' within union field '%s'.", type, unionField), "The following types are not supported: enum, map, array, and union.") .withConfigProperty(Conf.UNION_FIELD).withInputSchemaField(unionField); break; } String port = type == Schema.Type.RECORD ? schema.getRecordName() : type.name().toLowerCase(); outputFields.set(unionIndex, Schema.Field.of(unionField, modifySchema ? schema : unionSchema)); outputPortSchemas.put(port, Schema.recordOf(inputSchema.getRecordName() + "." + port, outputFields)); } return outputPortSchemas; } UnionSplitter(Conf conf); }
UnionSplitter extends SplitterTransform<StructuredRecord, StructuredRecord> { @VisibleForTesting static Map<String, Schema> getOutputSchemas(Schema inputSchema, String unionField, boolean modifySchema, FailureCollector collector) { Map<String, Schema> outputPortSchemas = new HashMap<>(); if (unionField == null) { outputPortSchemas.put(inputSchema.getRecordName(), inputSchema); return outputPortSchemas; } Schema.Field unionSchemaField = inputSchema.getField(unionField); if (unionSchemaField == null) { collector.addFailure(String.format("Field '%s' must exist in the input schema.", unionField), null) .withConfigProperty(Conf.UNION_FIELD); collector.getOrThrowException(); } Schema unionSchema = unionSchemaField.getSchema(); if (unionSchema.getType() != Schema.Type.UNION) { collector.addFailure(String.format("Field '%s' must be of type union, but is of type '%s'.", unionField, unionSchema.getType()), null) .withConfigProperty(Conf.UNION_FIELD); collector.getOrThrowException(); } int numFields = inputSchema.getFields().size(); ArrayList<Schema.Field> outputFields = new ArrayList<>(numFields); int i = 0; int unionIndex = -1; for (Schema.Field inputField : inputSchema.getFields()) { if (inputField.getName().equals(unionField)) { unionIndex = i; outputFields.add(null); } else { outputFields.add(inputField); } i++; } for (Schema schema : unionSchema.getUnionSchemas()) { Schema.Type type = schema.getType(); if (type.equals(Schema.Type.ENUM) || type.equals(Schema.Type.MAP) || type.equals(Schema.Type.ARRAY) || type.equals(Schema.Type.UNION)) { collector.addFailure(String.format("Unsupported type '%s' within union field '%s'.", type, unionField), "The following types are not supported: enum, map, array, and union.") .withConfigProperty(Conf.UNION_FIELD).withInputSchemaField(unionField); break; } String port = type == Schema.Type.RECORD ? schema.getRecordName() : type.name().toLowerCase(); outputFields.set(unionIndex, Schema.Field.of(unionField, modifySchema ? schema : unionSchema)); outputPortSchemas.put(port, Schema.recordOf(inputSchema.getRecordName() + "." + port, outputFields)); } return outputPortSchemas; } UnionSplitter(Conf conf); @Override void configurePipeline(MultiOutputPipelineConfigurer multiOutputPipelineConfigurer); @Override void transform(StructuredRecord record, MultiOutputEmitter<StructuredRecord> emitter); }
UnionSplitter extends SplitterTransform<StructuredRecord, StructuredRecord> { @VisibleForTesting static Map<String, Schema> getOutputSchemas(Schema inputSchema, String unionField, boolean modifySchema, FailureCollector collector) { Map<String, Schema> outputPortSchemas = new HashMap<>(); if (unionField == null) { outputPortSchemas.put(inputSchema.getRecordName(), inputSchema); return outputPortSchemas; } Schema.Field unionSchemaField = inputSchema.getField(unionField); if (unionSchemaField == null) { collector.addFailure(String.format("Field '%s' must exist in the input schema.", unionField), null) .withConfigProperty(Conf.UNION_FIELD); collector.getOrThrowException(); } Schema unionSchema = unionSchemaField.getSchema(); if (unionSchema.getType() != Schema.Type.UNION) { collector.addFailure(String.format("Field '%s' must be of type union, but is of type '%s'.", unionField, unionSchema.getType()), null) .withConfigProperty(Conf.UNION_FIELD); collector.getOrThrowException(); } int numFields = inputSchema.getFields().size(); ArrayList<Schema.Field> outputFields = new ArrayList<>(numFields); int i = 0; int unionIndex = -1; for (Schema.Field inputField : inputSchema.getFields()) { if (inputField.getName().equals(unionField)) { unionIndex = i; outputFields.add(null); } else { outputFields.add(inputField); } i++; } for (Schema schema : unionSchema.getUnionSchemas()) { Schema.Type type = schema.getType(); if (type.equals(Schema.Type.ENUM) || type.equals(Schema.Type.MAP) || type.equals(Schema.Type.ARRAY) || type.equals(Schema.Type.UNION)) { collector.addFailure(String.format("Unsupported type '%s' within union field '%s'.", type, unionField), "The following types are not supported: enum, map, array, and union.") .withConfigProperty(Conf.UNION_FIELD).withInputSchemaField(unionField); break; } String port = type == Schema.Type.RECORD ? schema.getRecordName() : type.name().toLowerCase(); outputFields.set(unionIndex, Schema.Field.of(unionField, modifySchema ? schema : unionSchema)); outputPortSchemas.put(port, Schema.recordOf(inputSchema.getRecordName() + "." + port, outputFields)); } return outputPortSchemas; } UnionSplitter(Conf conf); @Override void configurePipeline(MultiOutputPipelineConfigurer multiOutputPipelineConfigurer); @Override void transform(StructuredRecord record, MultiOutputEmitter<StructuredRecord> emitter);  }