@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 shouldUnionMapWithNulls() {
    final Map<String, String> input1 = Maps.newHashMap();
    input1.put("one", "apple");
    input1.put("two", "banana");
    input1.put("three", "cherry");

    final Map<String, String> input2 = Maps.newHashMap();
    input2.put("foo", "bar");
    input2.put(null, null);
    input2.put("baz", null);

    final Map<String, String> result = udf.union(input1, input2);
    assertThat(result.size(), is(6));
    assertThat(result.get("two"), is("banana"));
    assertThat(result.get("foo"), is("bar"));
    assertThat(result.get("baz"), is(nullValue()));
    assertThat(result.keySet(), containsInAnyOrder("one", "two", "three", null, "foo", "baz"));
  }