Pinot 1.1 Multi-stage query OOMs and brings down all servers -- with all the default protections/params in place

[dorlevi]

We have a realtime table (6 partitions, 140gb), when querying the table with timeout of 3 minutes all servers (6 servers, each 24cores and ~100 gb allocated to pinot) OOM and hangs

Query:

with dups as (
        select __global_counter,
                   __message_id,
                   __probe_id,
  min(__record_timestamp) as min_record_timestamp,
    max(__record_timestamp) as max_record_timestamp,
                  count(*) as cnt
        from org_2dYiMRMfas142XRKQ3bJIqmN3V6_ethereum_erc20_balance_changes_block_b91edb0804c04c2d9851eb975b689f0f
        group by __global_counter, __message_id, __probe_id
)

select 
*
from dups
order by cnt desc
limit 10

Explain plan:

Execution Plan
LogicalSort(sort0=[$5], dir0=[DESC], offset=[0], fetch=[10])
  PinotLogicalSortExchange(distribution=[hash], collation=[[5 DESC]], isSortOnSender=[false], isSortOnReceiver=[true])
    LogicalSort(sort0=[$5], dir0=[DESC], fetch=[10])
      LogicalAggregate(group=[{0, 1, 2}], agg#0=[MIN($3)], agg#1=[MAX($4)], agg#2=[COUNT($5)])
        PinotLogicalExchange(distribution=[hash[0, 1, 2]])
          LogicalAggregate(group=[{3, 4, 5}], agg#0=[MIN($6)], agg#1=[MAX($6)], agg#2=[COUNT()])
            LogicalTableScan(table=[[org_2dYiMRMfas142XRKQ3bJIqmN3V6_ethereum_erc20_balance_changes_block_b91edb0804c04c2d9851eb975b689f0f]])

We understand that such a query is perhaps not the best suited for Pinot but crashing all servers queried seems like a bug, especially as we haven't overridden any of the protections in place by the engine (besides timeout), we've reproduced it live for @mayankshriv and he suggested we open this issue.

OOM Logs from one of the servers (not super informative):

{"time":"2024-06-17T22:48:22.991521636+02:00","stream":"stdout","logtag":"F","message":"Terminating due to java.lang.OutOfMemoryError: Java heap space"}
--

Running server args (pinot 1.1):

\"args\" : [ \"--add-opens=java.base/java.nio=ALL-UNNAMED\", \"--add-opens=java.base/sun.nio.ch=ALL-UNNAMED\", \"--add-opens=java.base/java.lang=ALL-UNNAMED\", \"--add-opens=java.base/java.util=ALL-UNNAMED\", \"--add-opens=java.base/java.lang.reflect=ALL-UNNAMED\", \"-Xms32G\", \"-Xmx32G\", \"-XX:+ExitOnOutOfMemoryError\", \"-javaagent:/opt/pinot/etc/jmx_prometheus_javaagent/jmx_prometheus_javaagent.jar=8008:/opt/pinot/etc/jmx_prometheus_javaagent/configs/pinot.yml\", \"-Dlog4j2.configurationFile=/opt/pinot/etc/conf/pinot-server-log4j2.xml\", \"-Dplugins.dir=/opt/pinot/plugins\", \"-Dplugins.dir=/opt/pinot/plugins\", \"-Dapp.name=pinot-admin\", \"-Dapp.pid=1\", \"-Dapp.repo=/opt/pinot/lib\", \"-Dapp.home=/opt/pinot\", \"-Dbasedir=/opt/pinot\" ],"

Table config

{
  "REALTIME": {
    "tableName": "org_2dYiMRMfas142XRKQ3bJIqmN3V6_ethereum_erc20_balance_changes_block_b91edb0804c04c2d9851eb975b689f0f_REALTIME",
    "tableType": "REALTIME",
    "segmentsConfig": {
      "replication": "2",
      "retentionTimeUnit": "DAYS",
      "retentionTimeValue": "30",
      "replicasPerPartition": "1",
      "timeColumnName": "__record_timestamp",
      "minimizeDataMovement": false
    },
    "tenants": {
      "broker": "sim_community",
      "server": "sim_community"
    },
    "tableIndexConfig": {
      "columnMajorSegmentBuilderEnabled": false,
      "rangeIndexVersion": 2,
      "autoGeneratedInvertedIndex": false,
      "createInvertedIndexDuringSegmentGeneration": false,
      "sortedColumn": [
        "__global_counter",
        "__message_id"
      ],
      "loadMode": "MMAP",
      "enableDefaultStarTree": false,
      "enableDynamicStarTreeCreation": false,
      "aggregateMetrics": false,
      "nullHandlingEnabled": false,
      "optimizeDictionary": true,
      "optimizeDictionaryForMetrics": false,
      "noDictionarySizeRatioThreshold": 0.85
    },
    "metadata": {},
    "quota": {
      "maxQueriesPerSecond": "20.0"
    },
    "query": {
      "timeoutMs": 5000
    },
    "fieldConfigList": [
      {
        "name": "__global_counter",
        "encodingType": "RAW",
        "indexTypes": [],
        "indexes": null,
        "tierOverwrites": null
      }
    ],
    "ingestionConfig": {
      "streamIngestionConfig": {
        "streamConfigMaps": [
          {
            "metadata.populate": "true",
            "realtime.segment.flush.autotune.initialRows": "1000000",
            "realtime.segment.flush.threshold.rows": "0",
            "realtime.segment.flush.threshold.segment.size": "500M",
            "realtime.segment.serverUploadToDeepStore": "true",
            "sasl.jaas.config": "*****";",
            "sasl.mechanism": ****",
            "security.protocol": "******",
            "stream.kafka.broker.list": "*****",
            "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
            "stream.kafka.consumer.prop.auto.offset.reset": "smallest",
            "stream.kafka.consumer.type": "lowlevel",
            "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
            "stream.kafka.topic.name": "sim_f8ac4a56_17e7_4704_97a6_ba84f94b4d46_ethereum_erc20_balance_changes_block",
            "streamType": "kafka"
          }
        ],
        "columnMajorSegmentBuilderEnabled": false
      },
      "transformConfigs": [
        {
          "columnName": "__record_timestamp",
          "transformFunction": "__metadata$recordTimestamp"
        },
        {
          "columnName": "__global_counter",
          "transformFunction": "jsonPath(__internal, '$.global_counter')"
        },
        {
          "columnName": "__message_id",
          "transformFunction": "jsonPath(__internal, '$.message_id')"
        },
        {
          "columnName": "__probe_id",
          "transformFunction": "jsonPath(__internal, '$.probe_id')"
        },
        {
          "columnName": "txn_hash",
          "transformFunction": "concat('0x', jsonPath(__user, '$.txn_hash'))"
        },
        {
          "columnName": "block_number",
          "transformFunction": "jsonPath(__user, '$.block_number')"
        },
        {
          "columnName": "block_timestamp",
          "transformFunction": "jsonPath(__user, '$.block_timestamp')"
        },
        {
          "columnName": "token_address",
          "transformFunction": "concat('0x', jsonPath(__user, '$.token_address'))"
        },
        {
          "columnName": "token_name",
          "transformFunction": "jsonPath(__user, '$.token_name')"
        },
        {
          "columnName": "token_symbol",
          "transformFunction": "jsonPath(__user, '$.token_symbol')"
        },
        {
          "columnName": "token_decimals",
          "transformFunction": "bytesToBigDecimal(hexToBytes(concat('000000', jsonPath(__user, '$.token_decimals'))))"
        },
        {
          "columnName": "account_address",
          "transformFunction": "concat('0x', jsonPath(__user, '$.account_address'))"
        },
        {
          "columnName": "balance",
          "transformFunction": "bytesToBigDecimal(hexToBytes(concat('000000', jsonPath(__user, '$.balance'))))"
        }
      ],
      "continueOnError": false,
      "rowTimeValueCheck": true,
      "segmentTimeValueCheck": true
    },
    "isDimTable": false
  }
}

[gortiz]

Thanks for your report.

Apache Pinot has some OOM protection mechanisms that are applied to single-stage query engine but they are not applied in multi-stage query engine. I've created https://github.com/apache/pinot/issues/13436 in order to track them. We are actively working in some of them (specially in automatic query killing).

This specific query can be executed in single-stage. Have you try it there? Does it fail in that single-stage?

[dorlevi]

I need to find a proper time to test it because obviously killing our cluster is not something I can do at any time, (we don't have a test cluster with these big tables)

Regardless we care about the Multi-stage engine and we've seen it happen multiple times with different queries that some can only be executed on the multi-stage engine.

[mayankshriv]

cc: @Jackie-Jiang