@Override
    public BaseRow fillData(BaseRow input, Object line) {
        GenericRow genericRow = (GenericRow) input;
        Map<String, Object> cacheInfo = (Map<String, Object>) line;
        GenericRow row = new GenericRow(sideInfo.getOutFieldInfoList().size());
        row.setHeader(input.getHeader());
        String[] fields = sideInfo.getSideTableInfo().getFieldTypes();
        for (Map.Entry<Integer, Integer> entry: sideInfo.getInFieldIndex().entrySet()) {
            Object obj = genericRow.getField(entry.getValue());
            obj = convertTimeIndictorTypeInfo(entry.getValue(), obj);
            row.setField(entry.getKey(), obj);
        }

        for (Map.Entry<Integer, Integer> entry : sideInfo.getSideFieldIndex().entrySet()) {
            if (cacheInfo == null) {
                row.setField(entry.getKey(), null);
            } else {
                Object object = Es7Util.getTarget(cacheInfo.get(sideInfo.getSideFieldNameIndex().get(entry.getKey())), fields[entry.getValue()]);
                row.setField(entry.getKey(), object);
            }
        }

        return row;
    }
@description:
@program: flink.sql
@author: lany
@create: 2021/01/11 11:21
elasticsearch7/elasticsearch7-side/elasticsearch7-async-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch7/Elasticsearch7AsyncReqRow.java