@Udf
  public <T> Map<String, T> union(
      @UdfParameter(description = "first map to union") final Map<String, T> map1,
      @UdfParameter(description = "second map to union") final Map<String, T> map2) {

    final List<Map<String, T>> nonNullInputs =
        Stream.of(map1, map2)
        .filter(Objects::nonNull)
        .collect(Collectors.toList());
    if (nonNullInputs.size() == 0) {
      return null;
    }

    final Map<String, T> output = new HashMap<>();
    nonNullInputs
        .forEach(output::putAll);
    return output;
  }
@Test
  public void shouldHandleComplexValueTypes() {
    final Map<String, List<Double>> input1 = Maps.newHashMap();
    input1.put("apple", Arrays.asList(Double.valueOf(12.34), Double.valueOf(56.78)));
    input1.put("banana", Arrays.asList(Double.valueOf(43.21), Double.valueOf(87.65)));

    final Map<String, List<Double>> input2 = Maps.newHashMap();
    input2.put("foo", Arrays.asList(Double.valueOf(123.456)));

    final Map<String, List<Double>> result = udf.union(input1, input2);
    assertThat(result.size(), is(3));
    assertThat(result.get("banana"), contains(Double.valueOf(43.21), Double.valueOf(87.65)));
    assertThat(result.keySet(), containsInAnyOrder("foo", "banana", "apple"));
  }