Handle the elasticsearch exceptions in JDBC formatted outputs

[chloe-zh]

Issue #, if available:

• Issue Exception handling #320
• Issue "all shards fail" error when functions/operators are applied to null-value field #308

Description of changes:

The JDBC output and Json formatted result have different behaviors when exceptions are thrown from the ES engine. Typically these exceptions are due to bad requests, cluster issues and so forth. Currently we have tested out a couple type of such exceptions, the are either indeed or their root causes are ElasticsearchException instances.

It's unlikely to remodel JDBC driver to give the exactly same behavior with the ES raw response due to the limited JDBC types. This PR proposes to only change the behavior of the JDBC output and remain the Json output as it is looking like. The proposed behavior is looking like:
Query example:
SELECT CASE LOWER(OriginWeather) WHEN 'sunny' THEN '1' ELSE '2' END AS cases FROM kibana_sample_data_flights

/** Result from JDBC (assume there exists more than one shards and all shards failed): */
{
  "error": {
    "reason": "Error occurred in Elasticsearch engine: all shards failed",
    "details": """
Shard[0]: ScriptException[compile error]; nested: IllegalArgumentException[invalid sequence of tokens near ['lower_1'].]; nested: NoViableAltException;
Shard[1]: ...
Shard[2]: ...
...

For more details, please send request for Json format to see the raw response from elasticsearch engine.
""",
    "type": "SearchPhaseExecutionException"
  },
  "status": 400
}

---

Current behaviors look like:
Json format (thrown directly from the ES engine, which would be remained)

{
  "error": {
    "root_cause": [
      {
        "type": "script_exception",
        "reason": "compile error",
        "script_stack": [
          "if((def lower_1 = doc['OriginWeat ...",
          "        ^---- HERE"
        ],
        "script": "if((def lower_1 = doc['OriginWeather'].value.toLowerCase(Locale.forLanguageTag(\"en\"));lower_1 == 'sunny')){'1'} else {'2'}",
        "lang": "painless"
      }
    ],
    "type": "search_phase_execution_exception",
    "reason": "all shards failed",
    "phase": "query",
    "grouped": true,
    "failed_shards": [
      {
        "shard": 0,
        "index": "kibana_sample_data_flights",
        "node": "uGM73ev2RauYEnj6AboXyg",
        "reason": {
          "type": "script_exception",
          "reason": "compile error",
          "script_stack": [
            "if((def lower_1 = doc['OriginWeat ...",
            "        ^---- HERE"
          ],
          "script": "if((def lower_1 = doc['OriginWeather'].value.toLowerCase(Locale.forLanguageTag(\"en\"));lower_1 == 'sunny')){'1'} else {'2'}",
          "lang": "painless",
          "caused_by": {
            "type": "illegal_argument_exception",
            "reason": "invalid sequence of tokens near ['lower_1'].",
            "caused_by": {
              "type": "no_viable_alt_exception",
              "reason": null
            }
          }
        }
      }
    ]
  },
  "status": 400
}

JDBC output:

{
  "error": {
    "reason": "There was internal problem at backend",
    "details": "all shards failed",
    "type": "SearchPhaseExecutionException"
  },
  "status": 500
}

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

[penghuo]

Is it possible to merge with the existing bank index?

[chloe-zh]

I wonder if that will cause broken tests that use the bank index if inject null value into it.

[davidcui1225]

LGTM

[penghuo]

Thanks for the change!