@VisibleForTesting
  protected void flattenMap(GenericRow record, List<String> columns) {
    for (String column : columns) {
      Object value = record.getValue(column);
      if (value instanceof Map) {
        Map<String, Object> map = (Map) value;
        List<String> mapColumns = new ArrayList<>();
        for (Map.Entry<String, Object> entry : new ArrayList<>(map.entrySet())) {
          String flattenName = concat(column, entry.getKey());
          Object nestedValue = entry.getValue();
          record.putValue(flattenName, nestedValue);
          if (nestedValue instanceof Map || nestedValue instanceof Collection || isNonPrimitiveArray(nestedValue)) {
            mapColumns.add(flattenName);
          }
        }
        flattenMap(record, mapColumns);
      } else if (value instanceof Collection) {
        Collection collection = (Collection) value;
        if (_fieldsToUnnest.contains(column)) {
          for (Object inner : collection) {
            if (inner instanceof Map) {
              Map<String, Object> innerMap = (Map<String, Object>) inner;
              flattenMap(column, innerMap, new ArrayList<>(innerMap.keySet()));
            }
          }
        } else if (shallConvertToJson(collection)) {
          try {
            // convert the collection to JSON string
            String jsonString = JsonFunctions.jsonFormat(collection);
            record.putValue(column, jsonString);
          } catch (JsonProcessingException e) {
            throw new RuntimeException(
                String.format("Caught exception while converting value to JSON string %s", value), e);
          }
        }
      } else if (isNonPrimitiveArray(value)) {
        Object[] array = (Object[]) value;
        if (_fieldsToUnnest.contains(column)) {
          for (Object inner : array) {
            if (inner instanceof Map) {
              Map<String, Object> innerMap = (Map<String, Object>) inner;
              flattenMap(column, innerMap, new ArrayList<>(innerMap.keySet()));
            }
          }
        } else if (shallConvertToJson(array)) {
          try {
            // convert the array to JSON string
            String jsonString = JsonFunctions.jsonFormat(array);
            record.putValue(column, jsonString);
          } catch (JsonProcessingException e) {
            throw new RuntimeException(
                String.format("Caught exception while converting value to JSON string %s", value), e);
          }
        }
      }
    }
  }
@Test
  public void testFlattenMap() {
    ComplexTypeTransformer transformer = new ComplexTypeTransformer(new ArrayList<>(), ".");

    // test flatten root-level tuples
    GenericRow genericRow = new GenericRow();
    genericRow.putValue("a", 1L);
    Map<String, Object> map1 = new HashMap<>();
    genericRow.putValue("map1", map1);
    map1.put("b", "v");
    Map<String, Object> innerMap1 = new HashMap<>();
    innerMap1.put("aa", 2);
    innerMap1.put("bb", "u");
    innerMap1.put("cc", new byte[]{1, 1});

    map1.put("im1", innerMap1);
    Map<String, Object> map2 = new HashMap<>();
    map2.put("c", 3);
    genericRow.putValue("map2", map2);

    transformer.transform(genericRow);
    Assert.assertEquals(genericRow.getValue("a"), 1L);
    Assert.assertEquals(genericRow.getValue("map1.b"), "v");
    Assert.assertEquals(genericRow.getValue("map1.im1.aa"), 2);
    Assert.assertEquals(genericRow.getValue("map1.im1.bb"), "u");
    Assert.assertEquals(genericRow.getValue("map1.im1.cc"), new byte[]{1, 1});

    Assert.assertEquals(genericRow.getValue("map2.c"), 3);

    // test flattening the tuple inside the collection
    transformer = new ComplexTypeTransformer(Arrays.asList("l1"), ".");
    genericRow = new GenericRow();
    List<Map<String, Object>> list1 = new ArrayList<>();
    list1.add(map1);
    genericRow.putValue("l1", list1);
    List<Integer> list2 = new ArrayList<>();
    list2.add(2);
    genericRow.putValue("l2", list2);
    transformer.flattenMap(genericRow, new ArrayList<>(genericRow.getFieldToValueMap().keySet()));
    Map<String, Object> map = (Map<String, Object>) ((Collection) genericRow.getValue("l1")).iterator().next();
    Assert.assertEquals(map.get("b"), "v");
    Assert.assertEquals(map.get("im1.aa"), 2);
    Assert.assertEquals(map.get("im1.bb"), "u");

    // test overriding delimiter
    transformer = new ComplexTypeTransformer(Arrays.asList("l1"), "_");
    genericRow = new GenericRow();
    innerMap1 = new HashMap<>();
    innerMap1.put("aa", 2);
    innerMap1.put("bb", "u");
    map1 = new HashMap<>();
    map1.put("im1", innerMap1);
    list1 = new ArrayList<>();
    list1.add(map1);
    genericRow.putValue("l1", list1);
    transformer.flattenMap(genericRow, new ArrayList<>(genericRow.getFieldToValueMap().keySet()));
    map = (Map<String, Object>) ((Collection) genericRow.getValue("l1")).iterator().next();
    Assert.assertEquals(map.get("im1_aa"), 2);
    Assert.assertEquals(map.get("im1_bb"), "u");
  }