[BUG] PPL Join limitations

[noCharger]

What is the bug?

PPL join fails with circuit breaker exception even after adjusting circuit breaker settings. The join operation appears to be consuming excessive memory when trying to join large datasets on timestamp fields.

How can one reproduce the bug?
Steps to reproduce the behavior:

Enable calcite plugin:

curl -XPUT "http://localhost:9200/_cluster/settings" \
-H "Content-Type: application/json" \
-d'{
  "persistent": {
    "plugins.calcite.enabled": true
  }
}'

Increase circuit breaker limits to maximum:

curl -XPUT "http://localhost:9200/_cluster/settings" \
-H "Content-Type: application/json" \
-d'{
  "persistent": {
    "indices.breaker.fielddata.limit": "95%",
    "indices.breaker.total.limit": "95%",
    "indices.breaker.request.limit": "90%",
    "indices.breaker.fielddata.overhead": "1.03",
    "indices.breaker.request.overhead": "1.0"
  }
}'

Execute a simple JOIN query:

curl -XPOST "http://localhost:9200/_plugins/_ppl/" \
-H "Content-Type: application/json" \
-d'{
  "query": "source = big5 | left join on @timestamp = @timestamp [source = big5 | where `event.id` = '\''ERROR'\'' | stats count() by span(@timestamp, 1h)]"
}'

Observe the circuit breaker error:

{
  "error": {
    "reason": "Error occurred in OpenSearch engine: all shards failed",
    "details": "Shard[0]: OpenSearchException[java.util.concurrent.ExecutionException: CircuitBreakingException[[fielddata] Data too large, data for [_id] would be [859053756/819.2mb], which is larger than the limit of [858993459/819.1mb]]]",
    "type": "SearchPhaseExecutionException"
  },
  "status": 500
}

What is the expected behavior?

• Successfully execute join without hitting circuit breaker limits
• Handle large datasets efficiently
• Provide memory-efficient execution of joins

What is your host/environment?

• OS: Linux
• Version: 3.1
• Plugins

Do you have any additional context?
Using standard big5 index from OSB.

curl localhost:9200/_cat/indices
green open big5 Ta16685cTqeehNeEZ86wmw 1 0 116000000 0 25.8gb 25.8gb

[LantaoJin]

#3749 added a sub-search side logic limit for left join.

[LantaoJin]

@noCharger could you help to run explain for the issue query?

[LantaoJin]

data for [_id] would be [859053756/819.2mb] error message should be triggered by specific _id in query. （Field _id filled the data node memory, a possible case is sorting by _id). But the query in description seems not relevant, can you confirm the error is caused by this query? Again, the error is node level, multiple queries can be impacted by this error if a node reached circuit breaker limit. @noCharger

[noCharger]

Explain output for the same query, there's no _id presents in logical plan or physical plan

$curl -XPOST "localhost:9200/_plugins/_ppl/_explain?pretty" -H "Content-Type: application/json" -d'
{
  "query": "source = big5 | left join on @timestamp = @timestamp [source = big5 | where `event.id` = '\''ERROR'\'' | stats count() by span(@timest
amp, 1h)]"
}'
{
  "calcite": {
    "logical": "LogicalJoin(condition=[IS NOT NULL(TIMESTAMP($7))], joinType=[left])\n  LogicalProject(agent=[$0], process=[$6], log=[$8], message=[$11], tags=[$12], cloud=[$13], input=[$15], @timestamp=[$17], ecs=[$18], data_stream=[$20], meta=[$24], host=[$26], metrics=[$28], aws=[$31], event=[$36])\n    CalciteLogicalIndexScan(table=[[OpenSearch, big5]])\n  LogicalProject(count()=[$1], span(@timestamp,1h)=[$0])\n    LogicalAggregate(group=[{0}], count()=[COUNT()])\n      LogicalProject(span(@timestamp,1h)=[SPAN($17, 1, 'h')])\n        LogicalFilter(condition=[=($38, 'ERROR')])\n          CalciteLogicalIndexScan(table=[[OpenSearch, big5]])\n",
    "physical": "EnumerableCalc(expr#0..17=[{inputs}], proj#0..14=[{exprs}], count()=[$t16], span(@timestamp,1h)=[$t17])\n  EnumerableNestedLoopJoin(condition=[$15], joinType=[left])\n    EnumerableCalc(expr#0..14=[{inputs}], expr#15=[TIMESTAMP($t7)], expr#16=[IS NOT NULL($t15)], proj#0..14=[{exprs}], $f15=[$t16])\n      CalciteEnumerableIndexScan(table=[[OpenSearch, big5]], PushDownContext=[[PROJECT->[agent, process, log, message, tags, cloud, input, @timestamp, ecs, data_stream, meta, host, metrics, aws, event]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"agent\",\"process\",\"log\",\"message\",\"tags\",\"cloud\",\"input\",\"@timestamp\",\"ecs\",\"data_stream\",\"meta\",\"host\",\"metrics\",\"aws\",\"event\"],\"excludes\":[]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n    EnumerableCalc(expr#0..1=[{inputs}], count()=[$t1], span(@timestamp,1h)=[$t0])\n      EnumerableAggregate(group=[{0}], count()=[COUNT()])\n        EnumerableCalc(expr#0=[{inputs}], expr#1=[1], expr#2=['h'], expr#3=[SPAN($t0, $t1, $t2)], $f0=[$t3])\n          CalciteEnumerableIndexScan(table=[[OpenSearch, big5]], PushDownContext=[[PROJECT->[@timestamp, event.id], FILTER->=($1, 'ERROR'), PROJECT->[@timestamp]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"query\":{\"term\":{\"event.id\":{\"value\":\"ERROR\",\"boost\":1.0}}},\"_source\":{\"includes\":[\"@timestamp\"],\"excludes\":[]},\"sort\":[{\"_doc\":{\"order\":\"asc\"}}]}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
  }
}

@LantaoJin requestedTotalSize=2147483647 seems to be the issue.

[noCharger]

Known limitations:

1. head 10000 is the upper bound to avoid trigger CircuitBreakingException

curl -XPOST "localhost:9200/_plugins/_ppl" -H "Content-Type: application/json" -d'
{
  "query": "source = big5 | head 10000 | left join on @timestamp = @timestamp [source = big5 | where `event.id` = '\''ERROR'\'' | stats count() by span(@timestamp, 1h)]"
}'

2. reducing the matched doc set to 60K with filter getting same exception

$curl -XPOST "localhost:9200/_plugins/_ppl" -H "Content-Type: application/json" -d''
{
  "query": "source = big5 | where `@timestamp` >= '\''2023-01-01 00:00:00'\'' and `@timestamp` < '\''2023-01-01 00:10:00'\'' | stats count()"
}'
{
  "schema": [
    {
      "name": "count()",
      "type": "int"
    }
  ],
  "datarows": [
    [
      66532
    ]
  ],
  "total": 1,
  "size": 1
}

$ curl -XPOST "localhost:9200/_plugins/_ppl" -H "Content-Type: application/json" -d'
{
  "query": "source = big5 | where `@timestamp` >= '\''2023-01-01 00:00:00'\'' and `@timestamp` < '\''2023-01-01 00:10:00'\'' | left join on @timestamp = @timestamp [source = big5 | where `event.id` = '\''ERROR'\'' | stats count() by span(@timestamp, 1h)]"
}'
{
  "error": {
    "reason": "Error occurred in OpenSearch engine: all shards failed",
    "details": "Shard[0]: OpenSearchException[java.util.concurrent.ExecutionException: CircuitBreakingException[[fielddata] Data too large, data for [_id] would be [1021059953/973.7mb], which is larger than the limit of [1020054732/972.7mb]]]; nested: ExecutionException[CircuitBreakingException[[fielddata] Data too large, data for [_id] would be [1021059953/973.7mb], which is larger than the limit of [1020054732/972.7mb]]]; nested: CircuitBreakingException[[fielddata] Data too large, data for [_id] would be [1021059953/973.7mb], which is larger than the limit of [1020054732/972.7mb]];\n\nFor more details, please send request for Json format to see the raw response from OpenSearch engine.",
    "type": "SearchPhaseExecutionException"
  },
  "status": 500
}

[noCharger]

CircuitBreakingException resolved after changing the JVM heap size from 1G to 16G in OpenSearch. It's better to mention this on the public doc as well.

[LantaoJin]

Explain output for the same query, there's no _id presents in logical plan or physical plan

Yes, I knew, that's why I am confused about how data for [_id] would be be triggered.

@LantaoJin requestedTotalSize=2147483647 seems to be the issue.

IMO, it's not an issue, this is by design. Now we don't have any fetch limitation after #3623 merged: query.size_limit has been misused as maximum amount of for fetching documents from OpenSearch, causing many data correctness issues. It has now been fixed to the maximum number of rows output of query.