private GenericRow unnestCollection(GenericRow record, String column) {
    Object value = record.getValue(GenericRow.MULTIPLE_RECORDS_KEY);
    if (value == null) {
      List<GenericRow> list = new ArrayList<>();
      unnestCollection(record, column, list);
      record.putValue(GenericRow.MULTIPLE_RECORDS_KEY, list);
    } else {
      Collection<GenericRow> records = (Collection) value;
      List<GenericRow> list = new ArrayList<>();
      for (GenericRow innerRecord : records) {
        unnestCollection(innerRecord, column, list);
      }
      record.putValue(GenericRow.MULTIPLE_RECORDS_KEY, list);
    }
    return record;
  }
@Test
  public void testUnnestCollection() {
    // unnest root level collection
    //    {
    //      "array":[
    //      {
    //        "a":"v1"
    //      },
    //      {
    //        "a":"v2"
    //      }
    //   ]}
    //  ->
    //    [{
    //      "array.a":"v1"
    //    },
    //    {
    //      "array.a":"v2"
    //    }]
    ComplexTypeTransformer transformer = new ComplexTypeTransformer(Arrays.asList("array"), ".");
    GenericRow genericRow = new GenericRow();
    Object[] array = new Object[2];
    Map<String, Object> map1 = new HashMap<>();
    map1.put("a", "v1");
    Map<String, Object> map2 = new HashMap<>();
    map2.put("a", "v2");
    array[0] = map1;
    array[1] = map2;
    genericRow.putValue("array", array);
    transformer.transform(genericRow);
    Assert.assertNotNull(genericRow.getValue(GenericRow.MULTIPLE_RECORDS_KEY));
    Collection<GenericRow> collection = (Collection<GenericRow>) genericRow.getValue(GenericRow.MULTIPLE_RECORDS_KEY);
    Assert.assertEquals(2, collection.size());
    Iterator<GenericRow> itr = collection.iterator();
    Assert.assertEquals("v1", itr.next().getValue("array.a"));
    Assert.assertEquals("v2", itr.next().getValue("array.a"));

    // unnest sibling collections
    //    {
    //      "array":[
    //      {
    //        "a":"v1"
    //      },
    //      {
    //        "a":"v2"
    //      }],
    //      "array2":[
    //      {
    //        "b":"v3"
    //      },
    //      {
    //        "b":"v4"
    //      }]
    //    }
    // ->
    //  [
    //   {
    //      "array.a":"v1","array2.b":"v3"
    //   },
    //   {
    //      "array.a":"v1","array2.b":"v4"
    //   },
    //   {
    //      "array.a":"v2","array2.b":"v3"
    //   },
    //   {
    //      "array.a":"v2","array2.b":"v4"
    //   }]
    //
    transformer = new ComplexTypeTransformer(Arrays.asList("array", "array2"), ".");
    genericRow = new GenericRow();
    Object[] array2 = new Object[2];
    Map<String, Object> map3 = new HashMap<>();
    map3.put("b", "v3");
    Map<String, Object> map4 = new HashMap<>();
    map4.put("b", "v4");
    array2[0] = map3;
    array2[1] = map4;
    genericRow.putValue("array", array);
    genericRow.putValue("array2", array2);
    transformer.transform(genericRow);
    Assert.assertNotNull(genericRow.getValue(GenericRow.MULTIPLE_RECORDS_KEY));
    collection = (Collection<GenericRow>) genericRow.getValue(GenericRow.MULTIPLE_RECORDS_KEY);
    Assert.assertEquals(4, collection.size());
    itr = collection.iterator();
    GenericRow next = itr.next();
    Assert.assertEquals("v1", next.getValue("array.a"));
    Assert.assertEquals("v3", next.getValue("array2.b"));
    next = itr.next();
    Assert.assertEquals("v1", next.getValue("array.a"));
    Assert.assertEquals("v4", next.getValue("array2.b"));
    next = itr.next();
    Assert.assertEquals("v2", next.getValue("array.a"));
    Assert.assertEquals("v3", next.getValue("array2.b"));
    next = itr.next();
    Assert.assertEquals("v2", next.getValue("array.a"));
    Assert.assertEquals("v4", next.getValue("array2.b"));

    // unnest nested collection
    // {
    //   "array":[
    //      {
    //         "a":"v1",
    //         "array2":[
    //            {
    //               "b":"v3"
    //            },
    //            {
    //               "b":"v4"
    //            }
    //         ]
    //      },
    //      {
    //         "a":"v2",
    //         "array2":[
    //
    //         ]
    //      }
    //   ]}
    // ->
    // [
    //   {
    //      "array.a":"v1","array.array2.b":"v3"
    //   },
    //   {
    //      "array.a":"v1","array.array2.b":"v4"
    //   },
    //   {
    //      "array.a":"v2"
    //   }]
    transformer = new ComplexTypeTransformer(Arrays.asList("array", "array.array2"), ".");
    genericRow = new GenericRow();
    genericRow.putValue("array", array);
    map1.put("array2", array2);
    map2.put("array2", new Object[]{});
    transformer.transform(genericRow);
    Assert.assertNotNull(genericRow.getValue(GenericRow.MULTIPLE_RECORDS_KEY));
    collection = (Collection<GenericRow>) genericRow.getValue(GenericRow.MULTIPLE_RECORDS_KEY);
    Assert.assertEquals(3, collection.size());
    itr = collection.iterator();
    next = itr.next();
    Assert.assertEquals("v1", next.getValue("array.a"));
    Assert.assertEquals("v3", next.getValue("array.array2.b"));
    next = itr.next();
    Assert.assertEquals("v1", next.getValue("array.a"));
    Assert.assertEquals("v4", next.getValue("array.array2.b"));
    next = itr.next();
    Assert.assertEquals("v2", next.getValue("array.a"));
  }