@Override
    public void flatMap(BaseRow value, Collector<BaseRow> out) throws Exception {
        GenericRow genericRow = (GenericRow) value;
        List<Integer> equalValIndex = sideInfo.getEqualValIndex();
        ArrayList<Object> inputParams = equalValIndex.stream()
                .map(genericRow::getField)
                .collect(Collectors.toCollection(ArrayList::new));

        if (inputParams.size() != equalValIndex.size() && sideInfo.getJoinType() == JoinType.LEFT) {
            BaseRow row = fillData(value, null);
            RowDataComplete.collectBaseRow(out, row);
            return;
        }

        String cacheKey = inputParams.stream()
                .map(String::valueOf)
                .collect(Collectors.joining("_"));

        List<Map<String, Object>> cacheList = cacheRef.get().get(cacheKey);
        if (CollectionUtils.isEmpty(cacheList) && sideInfo.getJoinType() == JoinType.LEFT) {
            BaseRow row = fillData(value, null);
            RowDataComplete.collectBaseRow(out, row);
        } else if (!CollectionUtils.isEmpty(cacheList)) {
            cacheList.forEach(one -> RowDataComplete.collectBaseRow(out, fillData(value, one)));
        }
    }
side operator with cache for all(period reload)
Date: 2018/11/26
Company: www.dtstack.com

@author maqi
rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/AbstractRdbAllReqRow.java