一对一联查

结构

//多表联查
db.TABLE_NAME.aggregate([
    {
        $lookup:{//连表
            from:'table0',    //被关联的表名
            localField:'localField',    //关联标识符    主动关联的表
            foreignField:'foreignField',    //关联标识符    被关联的表
            as:'table0'        //被关联的表的别名，下面涉及到被关联的表的操作，都用这个
        }
    },
    {
        $unwind:'$table0'//扁平化,将数组数据拆分    被关联的表
    },
    {
        $match:{}//正常查询
        $match:{'or':[,]}//多条件查询    或
    },
    {
        $project:{                //要显示的字段
            '_id':0,    //_id
            'F1':'$table.f1',//取别称    主动关联的表的数据
            'F2':'$table0.f2',    被关联的表的数据
            'F3':{//case when
                $cond:{if:{$gte:['$f3',30]},then:0,else:50}
            }
        }
    },
]).forEach(function(item)){    //遍历    对多表联查出来的数据做操作    一般创建新表
    db.aaaa.insert(item);    //创建新表
}

案例

db.risk_group_info.aggregate([
    {
        '$lookup':{
            'from':'person',
            'localField':'person_id',
            'foreignField':'_id',
            'as':'person'
        }
    },
    {'$unwind':'$person'},
    {
        '$project':{
            'name':'$person.name'
         }
     }
])

结果

java代码

    Aggregation aggregation = Aggregation.newAggregation(
    Aggregation.lookup(MongoColConfig.COL_PERSON, "person_id", "_id", "person"),
    Aggregation.unwind("person", true),
    Aggregation.project("person.name").and("person.name").as("name"));
    AggregationResults aggregate = MongoTemplate.aggregate(aggregation, MongoColConfig.COL_RISK_GROUP_INFO, JSONObject.class);
    if (!ObjectUtils.isEmpty(aggregate) && MyCollectionUtil.isNotEmpty(aggregate.getMappedResults())) {
        List mappedResults = aggregate.getMappedResults();
    }

一对多联查

主表一条数据对应联查的副表的多条数据，主表对象中的集合存放副表的多条数据

mongo接收实体

@Data
public class SendAssistBaseJointQuery {
    private String id;
 
    private String primaryId;
 
    private String taskId;
 
    private String taskType;
 
    private String riskPersonType;
 
    private String caseId;
 
    private String caseName;
 
    private String pointPositionId;
 
    private String sitePointPositionName;
 
    private String pointPositionType;
 
    private String assistDataInfoId;
 
    private String cancelReason;
 
    private String backReason;
 
    private String backTime;
 
    private String dataSource;
 
    private String updateTime;
 
    // 任务表字段
    private String createTime;
    private String dealUsername;
    private String taskStatus;
    String sendAssistTaskStatus;
    String receiveFeedbackTaskStatus;
 
    private List assistDataInfos;
 
 
    @Data
    // assist_data_info表中字段
    public class AssistDataInfo {
        private String type;
        private String assistLetterName;
        private Integer sendCount;
        private Integer feedbackCount;
        private List assistDataAnnexes;
    }
}

java代码

        Criteria criteria = new Criteria();
        List aggregationOperation = new ArrayList<>();
 
        aggregationOperation.add(Aggregation.match(criteria));
        // 主表一对一，关联task表
        aggregationOperation.add(Aggregation.lookup(MongoColConfig.COL_TASK, "task_id", "_id", "task"));
        aggregationOperation.add(Aggregation.unwind("task", true));
        // 主表一对多，关联assist_data_info表
        aggregationOperation.add(Aggregation.lookup(MongoColConfig.ASSIST_DATA_INFO, "task_id", "task_id", "assist_data_info"));
        aggregationOperation.add(Aggregation.unwind("assist_data_info", true));
        aggregationOperation.add(Aggregation.project("_id")
                .and("_id").as("primary_id")
                .and("site_point_position_name").as("site_point_position_name")
                .and("task_id").as("task_id").and("task_type").as("task_type")
                .and("risk_person_type").as("risk_person_type").and("case_id").as("case_id")
                .and("case_name").as("case_name").and("point_position_id").as("point_position_id")
                .and("site_point_position_name").as("site_point_position_name").and("point_position_type").as("point_position_type")
                .and("assist_data_info_id").as("assist_data_info_id").and("cancel_reason").as("cancel_reason")
                .and("back_time").as("back_time")
                .and("data_source").as("data_source").and("update_time").as("update_time")
                .and("task.deal_username").as("deal_username").and("task.create_time").as("create_time")
                .and("task.task_status").as("task_status")
                .and("task.send_assist_task_status").as("send_assist_task_status")
                .and("task.receive_feedback_task_status").as("receive_feedback_task_status")
                .and("task.cancel_reason").as("cancel_reason")
                .and("task.back_reason").as("back_reason")
                .and("assist_data_info.type").as("type")
                .and("assist_data_info.assist_data_annexes").as("assist_data_annexes").and("assist_data_info.feedback_count").as("feedback_count")
                .and("assist_data_info.assist_letter_name").as("assist_letter_name").and("assist_data_info.send_count").as("send_count"));
 
 
        Aggregation aggregationPage = Aggregation.newAggregation(aggregationOperation);
 
        // 这些被basicDBObject.append的字段为assist_data_info表中字段。使用list集合接收
        BasicDBObject basicDBObject = new BasicDBObject();
        basicDBObject.append("type", "$type");
        basicDBObject.append("assist_data_annexes", "$assist_data_annexes");
        basicDBObject.append("assist_letter_name", "$assist_letter_name");
        basicDBObject.append("feedback_count", "$feedback_count");
        basicDBObject.append("send_count", "$send_count");
        // 根据task_id字段聚合，取每个字段第一个字段附别名。（这些是主字段）
        aggregationPage.getPipeline().add(Aggregation.group("task_id")
                .first("primary_id").as("primary_id")
                .first("site_point_position_name").as("site_point_position_name")
                .first("task_id").as("task_id").first("task_type").as("task_type")
                .first("risk_person_type").as("risk_person_type").first("case_id").as("case_id")
                .first("case_name").as("case_name").first("point_position_id").as("point_position_id")
                .first("site_point_position_name").as("site_point_position_name").first("point_position_type").as("point_position_type")
                .first("assist_data_info_id").as("assist_data_info_id").first("cancel_reason").as("cancel_reason")
                .first("back_reason").as("back_reason").first("back_time").as("back_time")
                .first("data_source").as("data_source").first("update_time").as("update_time")
                .first("deal_username").as("deal_username").first("create_time").as("create_time")
                .first("cancel_reason").as("cancel_reason")
                .first("task_status").as("task_status")
                .first("send_assist_task_status").as("send_assist_task_status")
                .first("receive_feedback_task_status").as("receive_feedback_task_status")
                // 需要和实体中assistDataInfos对应，集合名接收关联assist_data_info表中数据
                .addToSet(basicDBObject).as("assist_data_infos"));
 
        // 分页
        int skip = (pageReq.getPage() - 1) * pageReq.getPageSize();
        aggregationPage.getPipeline().add(Aggregation.sort(Sort.Direction.DESC, "create_time"));
        aggregationPage.getPipeline().add(Aggregation.skip((long) skip));
        aggregationPage.getPipeline().add(Aggregation.limit(ret.getPageSize()));
 
        aggregationPage = aggregationPage.withOptions(AggregationOptions.builder().allowDiskUse(true).build());
        AggregationResults aggregatePage = MongoTemplate.aggregate(aggregationPage, MongoColConfig.SEND_ASSIST_BASE, SendAssistBaseJointQuery.class);
        if (!ObjectUtils.isEmpty(aggregatePage) && MyCollectionUtil.isNotEmpty(aggregatePage.getMappedResults())) {
            List mappedResults = aggregatePage.getMappedResults();
        }
        // 总数
        aggregationTotal.getPipeline().add(Aggregation.group("task_id").count().as("count"));
        aggregationPage = aggregationPage.withOptions(AggregationOptions.builder().allowDiskUse(true).build());
        aggregationTotal = aggregationTotal.withOptions(AggregationOptions.builder().allowDiskUse(true).build());
        AggregationResults aggregateTotal = MongoTemplate.aggregate(aggregationTotal, MongoColConfig.SEND_ASSIST_BASE, JSONObject.class);
        if (!ObjectUtils.isEmpty(aggregateTotal) && MyCollectionUtil.isNotEmpty(aggregateTotal.getMappedResults())) {
            aggregateTotal.getMappedResults().size();
        }