ConcatDistinct implements AggregateFunction<String, ConcatDistinct> { public ConcatDistinct(String fieldName, Schema fieldSchema) { this.fieldName = fieldName; this.fieldSchema = fieldSchema; Type inputType = fieldSchema.isNullable() ? fieldSchema.getNonNullable().getType() : fieldSchema.getType(); if (!inputType.equals(Type.STRING)) { throw new IllegalArgumentException( String.format("Field '%s' is of unsupported non-string type '%s'. ", fieldName, inputType)); } } ConcatDistinct(String fieldName, Schema fieldSchema); @Override void initialize(); @Override void mergeValue(StructuredRecord record); @Override void mergeAggregates(ConcatDistinct otherAgg); @Override String getAggregate(); @Override Schema getOutputSchema();  }
@Test public void testConcatDistinct() { Schema fieldSchema = Schema.of(Schema.Type.STRING); Schema schema = Schema.recordOf("test", Schema.Field.of("x", Schema.nullableOf(fieldSchema))); test(new Concat("x", fieldSchema), schema, "x", "1, 2, 3, 4, 5", Arrays.asList("1", "2", "3", "4", "5"), new Concat("x", fieldSchema)); test(new Concat("x", fieldSchema), schema, "x", "2, 3, 4, 5", Arrays.asList(null, "2", "3", "4", "5"), new Concat("x", fieldSchema)); }