class JsonRowSerializationSchema implements SerializationSchema<Row> {

	private JsonNode convert(ContainerNode<?> container, JsonNode reuse, TypeInformation<?> info, Object object) {
		if (info == Types.VOID || object == null) {
			return container.nullNode();
		} else if (info == Types.BOOLEAN) {
			return container.booleanNode((Boolean) object);
		} else if (info == Types.STRING) {
			return container.textNode((String) object);
		} else if (info == Types.BIG_DEC) {
			// convert decimal if necessary
			if (object instanceof BigDecimal) {
				return container.numberNode((BigDecimal) object);
			}
			return container.numberNode(BigDecimal.valueOf(((Number) object).doubleValue()));
		} else if (info == Types.BIG_INT) {
			// convert integer if necessary
			if (object instanceof BigInteger) {
				return container.numberNode((BigInteger) object);
			}
			return container.numberNode(BigInteger.valueOf(((Number) object).longValue()));
		} else if (info == Types.SQL_DATE) {
			return container.textNode(object.toString());
		} else if (info == Types.SQL_TIME) {
			final Time time = (Time) object;
			// strip milliseconds if possible
			if (time.getTime() % 1000 > 0) {
				return container.textNode(timeFormatWithMillis.format(time));
			}
			return container.textNode(timeFormat.format(time));
		} else if (info == Types.SQL_TIMESTAMP) {
			return container.textNode(timestampFormat.format((Timestamp) object));
		} else if (info instanceof RowTypeInfo) {
			if (reuse != null && reuse instanceof ObjectNode) {
				return convertRow((ObjectNode) reuse, (RowTypeInfo) info, (Row) object);
			} else {
				return convertRow(null, (RowTypeInfo) info, (Row) object);
			}
		} else if (info instanceof ObjectArrayTypeInfo) {
			if (reuse != null && reuse instanceof ArrayNode) {
				return convertObjectArray((ArrayNode) reuse, ((ObjectArrayTypeInfo) info).getComponentInfo(), (Object[]) object);
			} else {
				return convertObjectArray(null, ((ObjectArrayTypeInfo) info).getComponentInfo(), (Object[]) object);
			}
		} else if (info instanceof BasicArrayTypeInfo) {
			if (reuse != null && reuse instanceof ArrayNode) {
				return convertObjectArray((ArrayNode) reuse, ((BasicArrayTypeInfo) info).getComponentInfo(), (Object[]) object);
			} else {
				return convertObjectArray(null, ((BasicArrayTypeInfo) info).getComponentInfo(), (Object[]) object);
			}
		} else if (info instanceof PrimitiveArrayTypeInfo && ((PrimitiveArrayTypeInfo) info).getComponentType() == Types.BYTE) {
			return container.binaryNode((byte[]) object);
		} else {
			// for types that were specified without JSON schema
			// e.g. POJOs
			try {
				return mapper.valueToTree(object);
			} catch (IllegalArgumentException e) {
				throw new IllegalStateException("Unsupported type information '" + info + "' for object: " + object, e);
			}
		}
	}

	public  JsonRowSerializationSchema(TypeInformation<Row> typeInfo);
	public  JsonRowSerializationSchema(String jsonSchema);

}