Select { public static <T> Flattened<T> flattenedSchema() { return new AutoValue_Select_Flattened.Builder<T>() .setNameFn(CONCAT_FIELD_NAMES) .setNameOverrides(Collections.emptyMap()) .build(); }  static Fields<T> create(); static Fields<T> fieldIds(Integer... ids); static Fields<T> fieldNames(String... names); static Fields<T> fieldAccess(FieldAccessDescriptor fieldAccessDescriptor); static Flattened<T> flattenedSchema();  }
@Test @Category(NeedsRunner.class) public void testFlatSchema() { List<Row> rows = IntStream.rangeClosed(0, 2) .mapToObj(i -> Row.withSchema(SIMPLE_SCHEMA).addValues(i, Integer.toString(i)).build()) .collect(Collectors.toList()); PCollection<Row> unnested = pipeline .apply(Create.of(rows).withRowSchema(SIMPLE_SCHEMA)) .apply(Select.flattenedSchema()); PAssert.that(unnested).containsInAnyOrder(rows); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testSimpleFlattening() { List<Row> bottomRow = IntStream.rangeClosed(0, 2) .mapToObj(i -> Row.withSchema(SIMPLE_SCHEMA).addValues(i, Integer.toString(i)).build()) .collect(Collectors.toList()); List<Row> rows = bottomRow.stream() .map(r -> Row.withSchema(NESTED_SCHEMA).addValues(r, r).build()) .collect(Collectors.toList()); PCollection<Row> unnested = pipeline .apply(Create.of(rows).withRowSchema(NESTED_SCHEMA)) .apply(Select.flattenedSchema()); assertEquals(UNNESTED_SCHEMA, unnested.getSchema()); List<Row> expected = bottomRow.stream() .map( r -> Row.withSchema(UNNESTED_SCHEMA) .addValues(r.getValue(0), r.getValue(1), r.getValue(0), r.getValue(1)) .build()) .collect(Collectors.toList()); ; PAssert.that(unnested).containsInAnyOrder(expected); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testAlternateNamePolicyFlatten() { List<Row> bottomRow = IntStream.rangeClosed(0, 2) .mapToObj(i -> Row.withSchema(SIMPLE_SCHEMA).addValues(i, Integer.toString(i)).build()) .collect(Collectors.toList()); List<Row> rows = bottomRow.stream() .map(r -> Row.withSchema(NESTED_SCHEMA2).addValues(r).build()) .collect(Collectors.toList()); PCollection<Row> unnested = pipeline .apply(Create.of(rows).withRowSchema(NESTED_SCHEMA2)) .apply(Select.<Row>flattenedSchema().keepMostNestedFieldName()); assertEquals(UNNESTED2_SCHEMA_ALTERNATE, unnested.getSchema()); List<Row> expected = bottomRow.stream() .map( r -> Row.withSchema(UNNESTED2_SCHEMA_ALTERNATE) .addValues(r.getValue(0), r.getValue(1)) .build()) .collect(Collectors.toList()); ; PAssert.that(unnested).containsInAnyOrder(expected); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testClashingNamePolicyFlatten() { List<Row> bottomRow = IntStream.rangeClosed(0, 2) .mapToObj(i -> Row.withSchema(SIMPLE_SCHEMA).addValues(i, Integer.toString(i)).build()) .collect(Collectors.toList()); thrown.expect(IllegalArgumentException.class); List<Row> rows = bottomRow.stream() .map(r -> Row.withSchema(NESTED_SCHEMA).addValues(r, r).build()) .collect(Collectors.toList()); PCollection<Row> unnested = pipeline .apply(Create.of(rows).withRowSchema(NESTED_SCHEMA)) .apply(Select.<Row>flattenedSchema().keepMostNestedFieldName()); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testClashingNameWithRenameFlatten() { List<Row> bottomRow = IntStream.rangeClosed(0, 2) .mapToObj(i -> Row.withSchema(SIMPLE_SCHEMA).addValues(i, Integer.toString(i)).build()) .collect(Collectors.toList()); List<Row> rows = bottomRow.stream() .map(r -> Row.withSchema(NESTED_SCHEMA).addValues(r, r).build()) .collect(Collectors.toList()); PCollection<Row> unnested = pipeline .apply(Create.of(rows).withRowSchema(NESTED_SCHEMA)) .apply( Select.<Row>flattenedSchema() .keepMostNestedFieldName() .withFieldNameAs("nested2.field1", "n2field1") .withFieldNameAs("nested2.field2", "n2field2")); assertEquals(CLASHING_NAME_UNNESTED_SCHEMA, unnested.getSchema()); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testFlattenWithOutputSchema() { List<Row> bottomRow = IntStream.rangeClosed(0, 2) .mapToObj(i -> Row.withSchema(SIMPLE_SCHEMA).addValues(i, Integer.toString(i)).build()) .collect(Collectors.toList()); List<Row> rows = bottomRow.stream() .map(r -> Row.withSchema(NESTED_SCHEMA).addValues(r, r).build()) .collect(Collectors.toList()); PCollection<Row> unnested = pipeline .apply(Create.of(rows).withRowSchema(NESTED_SCHEMA)) .apply(Select.<Row>flattenedSchema().withOutputSchema(CLASHING_NAME_UNNESTED_SCHEMA)); assertEquals(CLASHING_NAME_UNNESTED_SCHEMA, unnested.getSchema()); pipeline.run(); }