AddFields { public static <T> Inner<T> create() { return new Inner<>(); }  static Inner<T> create();  }
@Test @Category(NeedsRunner.class) public void addSimpleFields() { Schema schema = Schema.builder().addStringField("field1").build(); PCollection<Row> added = pipeline .apply( Create.of(Row.withSchema(schema).addValue("value").build()).withRowSchema(schema)) .apply( AddFields.<Row>create() .field("field2", Schema.FieldType.INT32) .field("field3", Schema.FieldType.array(Schema.FieldType.STRING)) .field("field4", Schema.FieldType.iterable(Schema.FieldType.STRING))); Schema expectedSchema = Schema.builder() .addStringField("field1") .addNullableField("field2", Schema.FieldType.INT32) .addNullableField("field3", Schema.FieldType.array(Schema.FieldType.STRING)) .addNullableField("field4", Schema.FieldType.iterable(Schema.FieldType.STRING)) .build(); assertEquals(expectedSchema, added.getSchema()); Row expected = Row.withSchema(expectedSchema).addValues("value", null, null, null).build(); PAssert.that(added).containsInAnyOrder(expected); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void addSimpleFieldsDefaultValue() { Schema schema = Schema.builder().addStringField("field1").build(); PCollection<Row> added = pipeline .apply( Create.of(Row.withSchema(schema).addValue("value").build()).withRowSchema(schema)) .apply(AddFields.<Row>create().field("field2", Schema.FieldType.INT32, 42)); Schema expectedSchema = Schema.builder() .addStringField("field1") .addField("field2", Schema.FieldType.INT32) .build(); assertEquals(expectedSchema, added.getSchema()); Row expected = Row.withSchema(expectedSchema).addValues("value", 42).build(); PAssert.that(added).containsInAnyOrder(expected); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void addNestedField() { Schema nested = Schema.builder().addStringField("field1").build(); Schema schema = Schema.builder().addRowField("nested", nested).build(); Row subRow = Row.withSchema(nested).addValue("value").build(); Row row = Row.withSchema(schema).addValue(subRow).build(); PCollection<Row> added = pipeline .apply(Create.of(row).withRowSchema(schema)) .apply( AddFields.<Row>create() .field("nested.field2", Schema.FieldType.INT32) .field("nested.field3", Schema.FieldType.array(Schema.FieldType.STRING)) .field("nested.field4", Schema.FieldType.iterable(Schema.FieldType.STRING))); Schema expectedNestedSchema = Schema.builder() .addStringField("field1") .addNullableField("field2", Schema.FieldType.INT32) .addNullableField("field3", Schema.FieldType.array(Schema.FieldType.STRING)) .addNullableField("field4", Schema.FieldType.iterable(Schema.FieldType.STRING)) .build(); Schema expectedSchema = Schema.builder().addRowField("nested", expectedNestedSchema).build(); assertEquals(expectedSchema, added.getSchema()); Row expectedNested = Row.withSchema(expectedNestedSchema).addValues("value", null, null, null).build(); Row expected = Row.withSchema(expectedSchema).addValue(expectedNested).build(); PAssert.that(added).containsInAnyOrder(expected); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void addNestedFieldDefaultValue() { Schema nested = Schema.builder().addStringField("field1").build(); Schema schema = Schema.builder().addRowField("nested", nested).build(); Row subRow = Row.withSchema(nested).addValue("value").build(); Row row = Row.withSchema(schema).addValue(subRow).build(); List<String> list = ImmutableList.of("one", "two", "three"); PCollection<Row> added = pipeline .apply(Create.of(row).withRowSchema(schema)) .apply( AddFields.<Row>create() .field("nested.field2", Schema.FieldType.INT32, 42) .field("nested.field3", Schema.FieldType.array(Schema.FieldType.STRING), list) .field( "nested.field4", Schema.FieldType.iterable(Schema.FieldType.STRING), list)); Schema expectedNestedSchema = Schema.builder() .addStringField("field1") .addField("field2", Schema.FieldType.INT32) .addField("field3", Schema.FieldType.array(Schema.FieldType.STRING)) .addField("field4", Schema.FieldType.iterable(Schema.FieldType.STRING)) .build(); Schema expectedSchema = Schema.builder().addRowField("nested", expectedNestedSchema).build(); assertEquals(expectedSchema, added.getSchema()); Row expectedNested = Row.withSchema(expectedNestedSchema).addValues("value", 42, list, list).build(); Row expected = Row.withSchema(expectedSchema).addValue(expectedNested).build(); PAssert.that(added).containsInAnyOrder(expected); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void addSimpleAndNestedField() { Schema nested = Schema.builder().addStringField("field1").build(); Schema schema = Schema.builder().addRowField("nested", nested).build(); Row subRow = Row.withSchema(nested).addValue("value").build(); Row row = Row.withSchema(schema).addValue(subRow).build(); PCollection<Row> added = pipeline .apply(Create.of(row).withRowSchema(schema)) .apply( AddFields.<Row>create() .field("field2", Schema.FieldType.INT32) .field("nested.field2", Schema.FieldType.INT32) .field("nested.field3", Schema.FieldType.array(Schema.FieldType.STRING)) .field("nested.field4", Schema.FieldType.iterable(Schema.FieldType.STRING))); Schema expectedNestedSchema = Schema.builder() .addStringField("field1") .addNullableField("field2", Schema.FieldType.INT32) .addNullableField("field3", Schema.FieldType.array(Schema.FieldType.STRING)) .addNullableField("field4", Schema.FieldType.iterable(Schema.FieldType.STRING)) .build(); Schema expectedSchema = Schema.builder() .addRowField("nested", expectedNestedSchema) .addNullableField("field2", Schema.FieldType.INT32) .build(); assertEquals(expectedSchema, added.getSchema()); Row expectedNested = Row.withSchema(expectedNestedSchema).addValues("value", null, null, null).build(); Row expected = Row.withSchema(expectedSchema).addValues(expectedNested, null).build(); PAssert.that(added).containsInAnyOrder(expected); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void recursivelyAddNestedFields() { Schema schema = Schema.of(); Row row = Row.withSchema(schema).build(); PCollection<Row> added = pipeline .apply(Create.of(row).withRowSchema(schema)) .apply( AddFields.<Row>create() .field("nested.field1", Schema.FieldType.STRING, "value") .field("nested.field2", Schema.FieldType.INT32) .field("nested.field3", Schema.FieldType.array(Schema.FieldType.STRING)) .field("nested.field4", Schema.FieldType.iterable(Schema.FieldType.STRING))); Schema expectedNestedSchema = Schema.builder() .addStringField("field1") .addNullableField("field2", Schema.FieldType.INT32) .addNullableField("field3", Schema.FieldType.array(Schema.FieldType.STRING)) .addNullableField("field4", Schema.FieldType.iterable(Schema.FieldType.STRING)) .build(); Schema expectedSchema = Schema.builder() .addNullableField("nested", Schema.FieldType.row(expectedNestedSchema)) .build(); assertEquals(expectedSchema, added.getSchema()); Row expectedNested = Row.withSchema(expectedNestedSchema).addValues("value", null, null, null).build(); Row expected = Row.withSchema(expectedSchema).addValue(expectedNested).build(); PAssert.that(added).containsInAnyOrder(expected); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void addNestedCollectionField() { Schema nested = Schema.builder().addStringField("field1").build(); Schema schema = Schema.builder() .addArrayField("array", Schema.FieldType.row(nested)) .addIterableField("iter", Schema.FieldType.row(nested)) .build(); Row subRow = Row.withSchema(nested).addValue("value").build(); Row row = Row.withSchema(schema) .addArray(subRow, subRow) .addIterable(ImmutableList.of(subRow, subRow)) .build(); PCollection<Row> added = pipeline .apply(Create.of(row).withRowSchema(schema)) .apply( AddFields.<Row>create() .field("array.field2", Schema.FieldType.INT32) .field("array.field3", Schema.FieldType.array(Schema.FieldType.STRING)) .field("iter.field2", Schema.FieldType.INT32) .field("iter.field3", Schema.FieldType.array(Schema.FieldType.STRING))); Schema expectedNestedSchema = Schema.builder() .addStringField("field1") .addNullableField("field2", Schema.FieldType.INT32) .addNullableField("field3", Schema.FieldType.array(Schema.FieldType.STRING)) .build(); Schema expectedSchema = Schema.builder() .addArrayField("array", Schema.FieldType.row(expectedNestedSchema)) .addIterableField("iter", Schema.FieldType.row(expectedNestedSchema)) .build(); assertEquals(expectedSchema, added.getSchema()); Row expectedNested = Row.withSchema(expectedNestedSchema).addValues("value", null, null).build(); Row expected = Row.withSchema(expectedSchema) .addArray(expectedNested, expectedNested) .addIterable(ImmutableList.of(expectedNested, expectedNested)) .build(); PAssert.that(added).containsInAnyOrder(expected); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void recursivelyAddNestedArrayField() { Schema schema = Schema.builder().build(); Row row = Row.withSchema(schema).build(); PCollection<Row> added = pipeline .apply(Create.of(row).withRowSchema(schema)) .apply( AddFields.<Row>create() .field("array[].field1", FieldType.STRING) .field("array[].field2", Schema.FieldType.INT32) .field("array[].field3", Schema.FieldType.array(Schema.FieldType.STRING))); Schema expectedNestedSchema = Schema.builder() .addNullableField("field1", FieldType.STRING) .addNullableField("field2", Schema.FieldType.INT32) .addNullableField("field3", Schema.FieldType.array(Schema.FieldType.STRING)) .build(); Schema expectedSchema = Schema.builder() .addNullableField( "array", Schema.FieldType.array( Schema.FieldType.row(expectedNestedSchema).withNullable(true))) .build(); assertEquals(expectedSchema, added.getSchema()); Row expected = Row.withSchema(expectedSchema).addValue(Collections.emptyList()).build(); PAssert.that(added).containsInAnyOrder(expected); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void addNestedMapField() { Schema nested = Schema.builder().addStringField("field1").build(); Schema schema = Schema.builder() .addMapField("map", Schema.FieldType.STRING, Schema.FieldType.row(nested)) .build(); Row subRow = Row.withSchema(nested).addValue("value").build(); Row row = Row.withSchema(schema).addValue(ImmutableMap.of("key", subRow)).build(); PCollection<Row> added = pipeline .apply(Create.of(row).withRowSchema(schema)) .apply( AddFields.<Row>create() .field("map.field2", Schema.FieldType.INT32) .field("map.field3", Schema.FieldType.array(Schema.FieldType.STRING)) .field("map.field4", Schema.FieldType.iterable(Schema.FieldType.STRING))); Schema expectedNestedSchema = Schema.builder() .addStringField("field1") .addNullableField("field2", Schema.FieldType.INT32) .addNullableField("field3", Schema.FieldType.array(Schema.FieldType.STRING)) .addNullableField("field4", Schema.FieldType.iterable(Schema.FieldType.STRING)) .build(); Schema expectedSchema = Schema.builder() .addMapField("map", Schema.FieldType.STRING, Schema.FieldType.row(expectedNestedSchema)) .build(); assertEquals(expectedSchema, added.getSchema()); Row expectedNested = Row.withSchema(expectedNestedSchema).addValues("value", null, null, null).build(); Row expected = Row.withSchema(expectedSchema).addValue(ImmutableMap.of("key", expectedNested)).build(); PAssert.that(added).containsInAnyOrder(expected); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void addDuplicateField() { Schema schema = Schema.builder().addStringField("field1").build(); thrown.expect(IllegalArgumentException.class); pipeline .apply(Create.of(Row.withSchema(schema).addValue("value").build()).withRowSchema(schema)) .apply(AddFields.<Row>create().field("field1", Schema.FieldType.INT32)); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void addNonNullableField() { Schema schema = Schema.builder().addStringField("field1").build(); thrown.expect(IllegalArgumentException.class); pipeline .apply(Create.of(Row.withSchema(schema).addValue("value").build()).withRowSchema(schema)) .apply(AddFields.<Row>create().field("field2", Schema.FieldType.INT32, null)); pipeline.run(); }