@Override
  public Iterable<GenericRow> transform(
      final K readOnlyKey,
      final GenericRow value,
      final KsqlProcessingContext ctx
  ) {
    if (value == null) {
      return null;
    }

    final List<Iterator<?>> iters = new ArrayList<>(tableFunctionAppliers.size());
    int maxLength = 0;
    for (final TableFunctionApplier applier : tableFunctionAppliers) {
      final List<?> exploded = applier.apply(value, processingLogger);
      iters.add(exploded.iterator());
      maxLength = Math.max(maxLength, exploded.size());
    }

    final List<GenericRow> rows = new ArrayList<>(maxLength);
    for (int i = 0; i < maxLength; i++) {
      final GenericRow newRow = new GenericRow(value.values().size() + iters.size());
      newRow.appendAll(value.values());

      for (final Iterator<?> iter : iters) {
        if (iter.hasNext()) {
          newRow.append(iter.next());
        } else {
          newRow.append(null);
        }
      }
      rows.add(newRow);
    }
    return rows;
  }
@Test
  public void shouldZipTwoFunctions() {
    // Given:
    final TableFunctionApplier applier1 = createApplier(Arrays.asList(10, 10, 10));
    final TableFunctionApplier applier2 = createApplier(Arrays.asList(20, 20));
    final KudtfFlatMapper<String> flatMapper =
        new KudtfFlatMapper<>(ImmutableList.of(applier1, applier2), processingLogger);

    // When:
    final Iterable<GenericRow> iterable = flatMapper.transform(KEY, VALUE, ctx);

    // Then:
    final Iterator<GenericRow> iter = iterable.iterator();
    assertThat(iter.next().values(), is(Arrays.asList(1, 2, 3, 10, 20)));
    assertThat(iter.next().values(), is(Arrays.asList(1, 2, 3, 10, 20)));
    assertThat(iter.next().values(), is(Arrays.asList(1, 2, 3, 10, null)));
    assertThat(iter.hasNext(), is(false));
  }