Elasticsearch merging documents in response

You cannot do query-time joins in Elasticsearch and will need to denormalize your data in order to efficiently retrieve and group it.

Having said that, you could:

1. leverage the multi-target syntax and query multiple indices at once
2. use an OR query on the id and userId -- since either of those is referenced at least once in any of your docs
3. and then trivially join your data through a map/reduce tool called scripted metric aggregations

Quick side note: you won't be able to use the _id keyword inside your docs because it's reserved.

Assuming your docs and indices are structured as follows:

POST users_index/_doc
{"id":"userId1","name":"John"}

POST invoices_index/_doc
{"id":"invoiceId1","userId":"userId1","cost":"10000","startdate":"","enddate":""}

POST orders_index/_doc
{"id":"orderId1","userId":"userId1","productName":"Mobile"}

Here's how the scripted metric aggregation could look like:

POST users_index,invoices_index,orders_index/_search
{
  "size": 0, 
  "query": {
    "bool": {
      "should": [
        {
          "term": {
            "id.keyword": {
              "value": "userId1"
            }
          }
        },
        {
          "term": {
            "userId.keyword": {
              "value": "userId1"
            }
          }
        }
      ]
    }
  },
  "aggs": {
    "group_by_invoiceId": {
      "scripted_metric": {
        "init_script": "state.users = []; state.invoices = []; state.orders = []",
        "map_script": """
          def source = params._source;
          
          if (source.containsKey("name")) {
            // we're dealing with the users index
            state.users.add(source);
          } else if (source.containsKey("cost")) {
            // we're dealing with the invoices index
            state.invoices.add(source);
          } else if (source.containsKey("productName")) {
            // we're dealing with the orders index
            state.orders.add(source);
          }
        """,
        "combine_script": """
          def non_empty_state = [:];
          for (entry in state.entrySet()) {
            if (entry != null && entry.getValue().length > 0) {
              non_empty_state[entry.getKey()] = entry.getValue();
            }
          }
          return non_empty_state;
        """,
        "reduce_script": """
          def final_invoices = [];
          
          def all_users = [];
          def all_invoices = [];
          def all_orders = [];
          
          // flatten all resources
          for (state in states) {
            for (kind_entry in state.entrySet()) {
              def map_kind = kind_entry.getKey();
              if (map_kind == "users") {
                all_users.addAll(kind_entry.getValue());
              } else if (map_kind == "invoices") {
                all_invoices.addAll(kind_entry.getValue());
              } else if (map_kind == "orders") {
                all_orders.addAll(kind_entry.getValue());
              } 
            }
          }
          
          // iterate the invoices and enrich them
          for (invoice_entry in all_invoices) {
            def invoiceId = invoice_entry.id;
            def userId = invoice_entry.userId;
            def userName = all_users.stream().filter(u -> u.id == userId).findFirst().get().name;
            def productName = all_orders.stream().filter(o -> o.userId == userId).findFirst().get().productName;
            def cost = invoice_entry.cost;
            def startdate = invoice_entry.startdate;
            def enddate = invoice_entry.enddate;
            
            final_invoices.add([
              'id': invoiceId,
              'userName': userName,
              'productName': productName,
              'cost': cost,
              'startdate': startdate,
              'enddate': enddate
            ]);
          }
          
          return final_invoices;
        """
      }
    }
  }
}

which'd return

{
  ...
  "aggregations" : {
    "group_by_invoiceId" : {
      "value" : [
        {
          "cost" : "10000",
          "enddate" : "",
          "id" : "invoiceId1",
          "userName" : "John",
          "startdate" : "",
          "productName" : "Mobile"
        }
      ]
    }
  }
}

Summing up, there are workarounds to achieve query-time joins. At the same time, scripts like this shouldn't be used in production because they could take forever.

Instead, this aggregation should be emulated outside of Elasticsearch after the query resolves and returns the index-specific hits.

BTW — I set size: 0 to return just the aggregation results so increase this parameter if you want to get some actual hits.