@Override
    public void flatMap(BaseRow value, Collector<BaseRow> out) throws Exception {
        GenericRow genericRow = (GenericRow) value;
        List<Object> inputParams = Lists.newArrayList();
        for (Integer conValIndex : sideInfo.getEqualValIndex()) {
            Object equalObj = genericRow.getField(conValIndex);
            if (equalObj == null) {
                sendOutputRow(value, null, out);
                return;
            }
            inputParams.add(equalObj);
        }

        String key = buildKey(inputParams);
        List<Map<String, Object>> cacheList = cacheRef.get().get(key);
        if (CollectionUtils.isEmpty(cacheList)) {
            sendOutputRow(value, null, out);
            return;
        }

        for (Map<String, Object> one: cacheList) {
            sendOutputRow(value, one, out);
            return;
        }

        for (Map<String, Object> one : cacheList) {
            sendOutputRow(value, one, out);
        }
    }
@description:
@program: flink.sql
@author: lany
@create: 2021/01/11 14:53
elasticsearch7/elasticsearch7-side/elasticsearch7-all-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch7/Elasticsearch7AllReqRow.java