code: private static CsvSchema.ColumnType convertType(String fieldName, TypeInformation<?> info) {
  		if (STRING_TYPES.contains(info)) {
  			return CsvSchema.ColumnType.STRING;
  		} else if (NUMBER_TYPES.contains(info)) {
  			return CsvSchema.ColumnType.NUMBER;
  		} else if (BOOLEAN_TYPES.contains(info)) {
  			return CsvSchema.ColumnType.BOOLEAN;
  		} else if (info instanceof ObjectArrayTypeInfo) {
  			validateNestedField(fieldName, ((ObjectArrayTypeInfo) info).getComponentInfo());
  			return CsvSchema.ColumnType.ARRAY;
  		} else if (info instanceof BasicArrayTypeInfo) {
  			validateNestedField(fieldName, ((BasicArrayTypeInfo) info).getComponentInfo());
  			return CsvSchema.ColumnType.ARRAY;
  		} else if (info instanceof RowTypeInfo) {
  			final TypeInformation<?>[] types = ((RowTypeInfo) info).getFieldTypes();
  			for (TypeInformation<?> type : types) {
  				validateNestedField(fieldName, type);
  			}
  			return CsvSchema.ColumnType.ARRAY;
  		} else if (info instanceof PrimitiveArrayTypeInfo &&
  				((PrimitiveArrayTypeInfo) info).getComponentType() == Types.BYTE) {
  			return CsvSchema.ColumnType.STRING;
  		} else {
  			throw new IllegalArgumentException(
  				"Unsupported type information '" + info.toString() + "' for field '" + fieldName + "'.");
  		}
  	}
docs: Convert {@link TypeInformation} to {@link CsvSchema.ColumnType} based on Jackson's categories.