[BUG] Cannot invoke \"java.util.Collection.size()\" because \"this.exprValues\" is null

[penghuo]

Issues

• statement state is update to SUCCESS. but result is EMPTY. Plugin can not handle this case now, will throw Exception. Cannot invoke \"java.util.Collection.size()\" because \"this.exprValues\" is null

Root cause

opensearch-project/sql#2436

Migration Plan

• Add sleep 2 seconds between write result index and update state.
  • choose 2 seconds because by default index refresh interval is 1s. add 1s as buffer.
  • sleep 2s does not impact current query latency.
  • sleep 2s increase query latency of query in waiting state. we does not expect much query in the queue, we accept this risk for now.
• Test with script and spark, 50 times, not BUG found.
• Spark, simulate EMR-S REPL job. https://github.com/penghuo/penghuo-opensearch-spark/blob/consistencyDebug/integ-test/src/test/scala/org/apache/spark/ConsistencyITSuite.scala#L37
• Python, simulate plugin REPL query handler

import requests
from requests.auth import HTTPBasicAuth
import json
import time
import threading
from datetime import datetime

# Elasticsearch settings
url = "https://xxxi.aos.us-west-2.on.aws"
es_user = ''
es_password = ''
es_index = 'query_execution_result_mys3'

def get_statement_state(docId):
    # print(f"[{docId}] read")
    es_url = f"{url}/.query_execution_request_mys3/_doc/{docId}"

    print(f'[{docId}] [{datetime.now()}] GET')
    response = requests.get(
        es_url,
        auth=HTTPBasicAuth(es_user, es_password),
        headers={"Content-Type": "application/json"},
    )
    if response.status_code == 200:
        result = response.json()
        print(f'[{docId}] [{datetime.now()}] current state: {result["_source"]["state"]}')
        return result["_source"]["state"]
    elif response.status_code == 404:
        # print(f"[{docId}] No statement doc found.")
        return "not_found"
    else:
        print(f"Failed to read documents. Status code: {response.status_code}")
        print(f"[{docId}] {response.text}")
        return "failed"


def search_result(docId):
    # print(f"[{docId}] search")
    es_url = f"{url}/query_execution_result_mys3/_search"
    query = {
              "query": {
              "bool": {
                "must": [
                  { "term": { "queryId": f"{docId}" } }
                ]
              }
            }}
    print(f'[{docId}] [{datetime.now()}] SEARCH')
    response = requests.get(
        es_url,
        auth=HTTPBasicAuth(es_user, es_password),
        headers={"Content-Type": "application/json"},
        json = query
    )
    if response.status_code == 200:
        result = response.json()
        hits_total_value = result["hits"]['total']['value']
        if (hits_total_value == 0):
            print(f"[{docId}] [{datetime.now()}] No result doc found.")
            return "not_found"
        elif hits_total_value == 1:
            print(f'[{docId}] Result doc. {result["hits"]["total"]}')
            return "success"
        else:
            print(f'[{docId}] Unexpected {result["hits"]}')
            return "failed"
    else:
        print(f"Failed to read documents. Status code: {response.status_code}")
        print(f"[{docId}] {response.text}")
        return "failed"

def repl(queryId):
  while True:
    result_state = search_result(queryId)
    st_state = get_statement_state(queryId)
    if (result_state == 'not_found') & (st_state == 'success'):
       print(f"[{queryId}] >>>>>>>>>> BUG FOUND >>>>>>>>>>")
    elif (result_state == 'success') & (st_state == 'success'):
       print(f"[{queryId}] SUCCESS")
       break
    time.sleep(1)

if __name__ == "__main__":
  # threads = []
  N = 50
  for i in range(1, 1 + N):
      repl(i)