BucketSelector pipeline aggregation extension

BucketSelector pipeline aggregation have some limitations -

1. Only one bucket selector can be applied on a parent multi-bucket aggregation. Since, BucketSelector retains the selected buckets and discards the others, so allowing multiple BucketSelector for an aggregation isn't the ideal behavior.
2. Key filters are not supported. Script variables in BucketSelector only pertains to the numeric values and keys cannot be used a script variable to write selection expression.
3. Doesn't works with composite aggregations: None of the pipeline aggregation works with composite aggregation as most of the pipeline aggregations works on entirety of the results, whereas, composite aggregations results could be paginated. Refer: https://github.com/elastic/elasticsearch/issues/32692

Proposal
BucketSelectorExt would be an extension to BucketSelector.

With BucketSelectorExt above limitations can be addressed -

• For 1, each BucketSelectorExt will have its own output section displaying the index of the selected buckets from the parent multi-bucket aggregation instead of the actual bucket. Also, the parent aggregation will contain all buckets and BucketSelectorExt will not have any impact of its result.
• For 2, there would be a optional filter field. Here one can pass an include/exclude filter which would works on the lines of term aggregation filtering supported by elasticsearch.
• For 3, key filters should support passing the filter for each source of the composite aggregation. For this, we have introduced a new key, value object where key is the name of the source and value is the key filter for the corresponding source in composite aggregation. Refer examples below.

Parameters:
parent_bucket_path - this is to navigate to the right parent multi-bucket aggregation on which selector has to be applied. It supports nested aggregations but should comply with below constraint -
agg1>agg2>agg3 - where agg1 and agg2 are all single-bucket aggs. Whereas, agg3 i.e. the last aggregation in the hierarchy should be a multi-bucket aggregation on which bucket selector would be applicable.
buckets_path - this is same as existing BucketSelector buckets_path

script - this is same as existing BucketSelector script

filter - key filter condition. First keys are filtered and then the bucket selector scripts are executed on the filtered keys.
It containsinclude/exclude filter which works on the lines of term aggregation filtering supported by elasticsearch.

composite_agg_filter: key filter condition for composite aggregations. Refer to example below for usage.

Some usage examples -

• Simple case -

"aggs": {
   "<parent_agg_name>": {
      "<child_multi_bucket_agg_name>": { "terms": {"field": "<fieldname>"}}},
      "aggs": {
         "<metric_agg_name>": { "stats": { "field": "<fieldname>" } }
      }
   },
   "<bucket_selector_name>": {
      "bucket_selector_ext": {
         "buckets_path": {
           "metric_value": "<metric_agg_name>.<metric_name>"
         },
         "script": {
           "source": "params.metric_value >= 10.0"
         },
         "parent_bucket_path": "<parent_agg_name>"
      }
   }
}

• Multiple bucket selectors selectors

"aggs": {
   "<parent_agg_name>": {
      "<child_multi_bucket_agg_name>": { "terms": {"field": "<fieldname>"}}},
      "aggs": {
         "<metric_agg_name>": { "stats": { "field": "<fieldname>" } }
      }
   },
   "<bucket_selector_name_1>": {
      "bucket_selector_ext": {
         "buckets_path": {
           "metric_value": "<metric_agg_name>.<metric_name>"
         },
         "script": {
           "source": "params.metric_value >= 10.0"
         },
         "parent_bucket_path": "<parent_agg_name>"
      }
   },
   "<bucket_selector_name_2>": {
      "bucket_selector_ext": {
         "buckets_path": {
           "metric_value": "<metric_agg_name>.<metric_name>"
         },
         "script": {
           "source": "params.metric_value >= 10.0"
         },
         "parent_bucket_path": "<parent_agg_name>"
      }
   }
}

• Key filters -

"aggs": {
   "<parent_agg_name>": {
      "<child_multi_bucket_agg_name>": { "terms": {"field": "<fieldname>"}}},
      "aggs": {
         "<metric_agg_name>": { "stats": { "field": "<fieldname>" } }
      }
   },
   "<bucket_selector_name>": {
      "bucket_selector_ext": {
         "buckets_path": {
           "metric_value": "<metric_agg_name>.<metric_name>"
         },
         "script": {
           "source": "params.metric_value >= 10.0"
         },
         "parent_bucket_path": "<parent_agg_name>",
         "filter": {
           "include": ["key1", "key2"]
         }
      }
   }
}

For regex, refer lucene regular expression

"aggs": {
   "<parent_agg_name>": {
      "<child_multi_bucket_agg_name>": { "terms": {"field": "<fieldname>"}}},
      "aggs": {
         "<metric_agg_name>": { "stats": { "field": "<fieldname>" } }
      }
   },
   "<bucket_selector_name_1>": {
      "bucket_selector_ext": {
         "buckets_path": {
           "metric_value": "<metric_agg_name>.<metric_name>"
         },
         "script": {
           "source": "params.metric_value >= 10.0"
         },
         "parent_bucket_path": "<parent_agg_name>",
         "filter": {
           "include": "key_prefix*"
         }
      }
   }
}

• Composite aggregation -

"aggs": {
   "<parent_agg_name>": {
      "composite": {
         "sources": [
            {"<source_1>": { "terms": {"field": "<field_1>"}}},
            {"<source_2>": { "terms": {"field": "<field_2>" }}}  
         ]
      },      
      "aggs": {
         "<metric_agg_name>": { "stats": { "field": "<fieldname>" } }
      }
   },
   "<bucket_selector_name_1>": {
      "bucket_selector_ext": {
         "buckets_path": {
           "metric_value": "<metric_agg_name>.<metric_name>"
         },
         "script": {
           "source": "params.metric_value >= 10.0"
         },
         "parent_bucket_path": "<parent_agg_name>",
         "composite_agg_filter": {
            "<source_1>" : {
               "include": ["<include_key_1>"]
            },
            "<source_2>" : {
               "include": "@"
            }
         }
      }
   }
}

We are planning to add this as part of alerting plugin - https://github.com/opendistro-for-elasticsearch/alerting/pull/374
This enhancement could be useful for others too and we can add to upstream as well. Currently, its implemented in kotlin, but I can rewrite in java if get enough votes.

Start by reviewing the existing BucketSelector aggregation and the alerting plugin implementation referenced in PR #374. Compare the proposed parent_bucket_path, filter, and composite_agg_filter behavior with the documented limitations, and confirm that the extension supports multiple selectors without changing the parent aggregation; the issue provides request examples for the expected behavior.