public static Builder forRow(DataStream<Row> input, TableSchema tableSchema) {
        RowType rowType = (RowType) tableSchema.toRowDataType().getLogicalType();
        DataType[] fieldDataTypes = tableSchema.getFieldDataTypes();

        DataFormatConverters.RowConverter rowConverter = new DataFormatConverters.RowConverter(fieldDataTypes);
        return builderFor(input, rowConverter::toInternal, FlinkCompatibilityUtil.toTypeInfo(rowType))
                .tableSchema(tableSchema);
    }
Initialize a {@link Builder} to export the data from input data stream with {@link Row}s into iceberg table.
We use {@link RowData} inside the sink connector, so users need to provide a {@link TableSchema} for builder to
convert those {@link Row}s to a {@link RowData} DataStream.

@param input       the source input data stream with {@link Row}s.
@param tableSchema defines the {@link TypeInformation} for input data.
@return {@link Builder} to connect the iceberg table.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java