private static RuntimeConverter assembleRowRuntimeConverter(
			boolean isTopLevel,
			String[] fieldNames,
			RuntimeConverter[] fieldConverters) {
		final int rowArity = fieldNames.length;
		// top level reuses the object node container
		if (isTopLevel) {
			return (csvMapper, container, obj) -> {
				final Row row = (Row) obj;

				validateArity(rowArity, row.getArity());

				final ObjectNode objectNode = (ObjectNode) container;
				for (int i = 0; i < rowArity; i++) {
					objectNode.set(
						fieldNames[i],
						fieldConverters[i].convert(csvMapper, container, row.getField(i)));
				}
				return objectNode;
			};
		} else {
			return (csvMapper, container, obj) -> {
				final Row row = (Row) obj;

				validateArity(rowArity, row.getArity());

				final ArrayNode arrayNode = csvMapper.createArrayNode();
				for (int i = 0; i < rowArity; i++) {
					arrayNode.add(fieldConverters[i].convert(csvMapper, arrayNode, row.getField(i)));
				}
				return arrayNode;
			};
		}
	}
Creates a {@link CsvRowSerializationSchema} expecting the given {@link TypeInformation}.

@param typeInfo type information used to create schema.
kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/serialization/CsvTupleSerializationSchema.java