基于query内的distinct如何翻译成DSL

基于query内的distinct如何翻译成DSL，ES版本6.7.1

如何将下列sql 翻译成ES 的DSL呢：
select hphm,hpzl,count(*) from (
select distinct hphm,hpzl, from its_carrecords
where rq='2020-12-01' and sj>='14:00:00' and sj<='15:00:00' and dwbh in ('10001','10002','5008')
union all
select distinct hphm,hpzl from its_carrecords
where rq='2020-12-01' and sj>='14:00:00' and sj<='15:00:00' and dwbh in ('10001','10002','5008')
union all
select distinct hphm,hpzl from its_carrecords
where rq='2020-12-01' and sj>='14:00:00' and sj<='15:00:00' and dwbh in ('10001','10002','5008')
)
group by hphm,hpzl having count(*)>2

我已经翻译成部分DSL如下：
GET /gcindex/_search
{
"size": 0,
"query": {
"bool": {
"should": [
{
"bool": {
"filter": [
{
"range": {
"passtime": {
"from": "2020-11-18 10:48:22.000",
"to": "2020-11-19 10:48:22.000"
}
}
},
{
"terms": {
"dwbh": [
"1165",
"1201",
"4810",
"184"
]
}
}
]
}
},
{
"bool": {
"filter": [
{
"range": {
"passtime": {
"from": "2020-11-18 10:49:07.000",
"to": "2020-11-18 10:49:07.000"
}
}
},
{
"terms": {
"dwbh": [
"1165",
"1201",
"4810",
"184"
]
}
}
]
}
}
]
}
},
"aggs": {
"grouphphms": {
"terms": {
"field": "hphm_key"
},
"aggs": {
"grouphpzl": {
"terms": {"field": "hpzl"}
},
"having": {
"bucket_selector": {
"buckets_path": {
"itemCount": "_count"
},
"script": {
"source": "params.itemCount >=1 "
}
}
}
}
}
}
}
问题是基于query的distinct如何翻译？
用cardinality是没用的，cardinality是基于聚合的distinct，我这个是query内部的distinct
用collapse也是不行的，collapse对聚合不影响，不是我要的效果。